fix(api): surface each device's own catalog category per-match

The per-device entries in an upload's matched_devices list were labelled
with the ML overall predicted_category, producing contradictions like
"Acurite Tower Sensor -> Fan Controller". The engine now preserves each
DeviceMatch's true catalog category in match_details["device_category"],
and the API surfaces that per device. The headline device_category still
reflects the ML overall call, so the honest gate-metric category is
unchanged.

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

.dockerignore

@@ -0,0 +1,36 @@
+# Keep the image small and reproducible. The 11 GB test corpora, the local
+# venv, and the vendored Flipper firmware are NOT needed to serve the app.
+
+# Heavy / irrelevant
+data/rf_test_datasets/
+data/test_db/
+data/test_known_devices/
+data/test_rtl433_real/
+signatures/
+venv/
+logs/
+docs/
+tests/
+test_files/
+scripts/
+
+# Benched Phase 3B CNN binaries (not on the serving path)
+models/category_cnn.pt
+models/category_cnn.onnx
+
+# VCS / caches / editor
+.git/
+.gitignore
+**/__pycache__/
+**/*.pyc
+**/*.pyo
+.pytest_cache/
+.mypy_cache/
+*.egg-info/
+.DS_Store
+
+# Docs & planning (not needed at runtime)
+*.md
+CONTEXT.md
+FABLE.md
+PLAN_TO_PROD.md

Dockerfile

@@ -0,0 +1,41 @@
+# GigLez — production image for the live "simple mode" API.
+# Serves: FastAPI upload/map UI + statistical device-category identification.
+FROM python:3.10-slim
+
+# Faster, quieter, unbuffered logs in containers.
+ENV PYTHONUNBUFFERED=1 \
+    PYTHONDONTWRITEBYTECODE=1 \
+    PIP_NO_CACHE_DIR=1
+
+WORKDIR /app
+
+# Dependencies first for layer caching — only re-runs when reqs change.
+COPY requirements-prod.txt ./
+RUN pip install --no-cache-dir -r requirements-prod.txt
+
+# Application code + assets the live path needs.
+COPY src/ ./src/
+COPY static/ ./static/
+COPY templates/ ./templates/
+COPY config/ ./config/
+COPY models/category_classifier.joblib ./models/category_classifier.joblib
+
+# Seed data: the static rtl_433 protocol table (read-only) and an initial
+# capture store. When /app/data is mounted as a *named volume*, Docker seeds
+# the empty volume from these baked files on first run, then persists writes
+# to captures_simple.json across restarts.
+COPY data/rtl_433_protocols.json data/captures_simple.json data/weather_sensors_found.txt ./data/
+
+# Run as a non-root user.
+RUN useradd --create-home --uid 10001 giglez \
+    && chown -R giglez:giglez /app
+USER giglez
+
+EXPOSE 8000
+
+# Container-native healthcheck hits the app's /health endpoint.
+HEALTHCHECK --interval=30s --timeout=3s --start-period=15s --retries=3 \
+    CMD python -c "import urllib.request,sys; sys.exit(0 if urllib.request.urlopen('http://127.0.0.1:8000/health').status==200 else 1)"
+
+# Serve with uvicorn (module app object), not the __main__ dev banner.
+CMD ["uvicorn", "src.api.main_simple:app", "--host", "0.0.0.0", "--port", "8000"]

docker-compose.yml

@@ -0,0 +1,22 @@
+# GigLez — single-service deployment of the live simple-mode API.
+#   docker compose up --build   → http://localhost:8000
+services:
+  giglez:
+    build: .
+    image: giglez:latest
+    ports:
+      - "8000:8000"
+    volumes:
+      # Persist the JSON capture store across restarts. Seeded from the
+      # baked-in data/ on first run (see Dockerfile), then app writes survive.
+      - giglez_data:/app/data
+    restart: unless-stopped
+    healthcheck:
+      test: ["CMD", "python", "-c", "import urllib.request,sys; sys.exit(0 if urllib.request.urlopen('http://127.0.0.1:8000/health').status==200 else 1)"]
+      interval: 30s
+      timeout: 3s
+      start_period: 15s
+      retries: 3
+
+volumes:
+  giglez_data:

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
+  }
+}

models/category_cnn_metrics.json

@@ -0,0 +1,36 @@
+{
+  "n_samples": 803,
+  "n_train": 489,
+  "n_val": 94,
+  "n_test": 220,
+  "classes": [
+    "Doorbell",
+    "Fan Controller",
+    "Garage Door Opener",
+    "Remote Control",
+    "Security Sensor",
+    "Weather Sensor"
+  ],
+  "pulse_len": 512,
+  "epochs": 60,
+  "class_balance": {
+    "Doorbell": 39,
+    "Garage Door Opener": 565,
+    "Weather Sensor": 12,
+    "Fan Controller": 91,
+    "Security Sensor": 2,
+    "Remote Control": 94
+  },
+  "cnn": {
+    "accuracy": 0.7954545454545454,
+    "balanced_accuracy": 0.33796618290289177,
+    "macro_f1": 0.3552945963506287
+  },
+  "heuristic_same_test": {
+    "top1": 0.3
+  },
+  "statistical_same_test": {
+    "top1": 0.8681818181818182,
+    "balanced_accuracy": 0.6253728690437551
+  }
+}

requirements-prod.txt

@@ -0,0 +1,30 @@
+# GigLez production runtime — the LIVE simple-mode path only
+# (src/api/main_simple.py → SignatureMatcher → pattern_decoder → statistical ML).
+#
+# Versions are pinned to what actually trained/serves the model in this env.
+# In particular scikit-learn/numpy MUST match the versions the
+# models/category_classifier.joblib bundle was built with (1.6.1 / 2.2.x),
+# or joblib.load() will warn/break. Do NOT downgrade to the old
+# requirements.txt pins — those drive the dormant SQLAlchemy/PostGIS path.
+#
+# NOTE: torch/onnx are intentionally absent — the Phase 3B CNN is benched
+# (loses to the statistical model), so it is not part of the serving path.
+
+# Web stack
+fastapi==0.121.1
+starlette==0.46.0
+uvicorn[standard]==0.31.1
+jinja2==3.1.6
+python-multipart==0.0.22
+pydantic==2.12.4
+
+# Numerics + statistical ML (RAW category classifier)
+numpy==2.2.6
+scipy==1.15.3
+scikit-learn==1.6.1
+joblib==1.5.3
+
+# Support
+loguru==0.7.2
+pyyaml==6.0.2
+python-dateutil==2.9.0.post0

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/experiment_rtl433_augment.py

@@ -0,0 +1,211 @@
+#!/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()

scripts/rtl433_iq_demod.py

@@ -0,0 +1,142 @@
+#!/usr/bin/env python3
+"""
+rtl_433 .cu8 IQ  ->  Flipper-style signed pulse train (µs)
+==========================================================
+
+Turns the rtl_433 project's ``.cu8`` test captures (interleaved I/Q uint8) into
+the same signed-duration pulse array our RAW feature extractor already consumes
+(``+high`` / ``-low`` in microseconds), so genuinely-independent, non-Flipper
+captures can augment the statistical category classifier's training set.
+
+Why amplitude (OOK) demod: the target Sub-GHz device classes we care about
+(weather sensors, doorbells, PIR/door/smoke security sensors, simple remotes)
+are overwhelmingly OOK/ASK — the carrier is keyed on/off, so the signal
+*envelope* (magnitude of the complex sample) is the message. FSK captures keep
+constant amplitude, so this demod yields no usable transitions on them; that is
+a feature, not a bug — ``demod_file`` returns ``None`` and such captures
+self-filter out of the training set via ``looks_like_ook``.
+
+The sample rate and centre frequency are read from the rtl_433 filename
+convention ``*_<freq>M_<rate>k.cu8`` (e.g. ``g003_433.92M_250k.cu8``), falling
+back to 250 kHz / 433.92 MHz.
+
+This is a training-data-prep tool (offline), not part of the serving path.
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+import numpy as np
+
+DEFAULT_RATE = 250_000
+DEFAULT_FREQ = 433_920_000
+
+# Quality gate — reject captures that did not demodulate into a plausible
+# OOK pulse train (FSK, pure noise, or empty).
+MIN_TRANSITIONS = 16      # need real structure, not one long burst
+MIN_SHORT_US = 40         # ignore sub-symbol glitches
+MAX_KEEP_PULSES = 1024    # cap runaway repeats (Flipper captures are bounded)
+
+
+def parse_rate_freq(name: str) -> Tuple[int, int]:
+    """Extract (sample_rate_hz, frequency_hz) from an rtl_433 filename."""
+    rate_m = re.search(r"_(\d+)k", name)
+    rate = int(rate_m.group(1)) * 1000 if rate_m else DEFAULT_RATE
+    freq_m = re.search(r"_(\d+(?:\.\d+)?)M", name)
+    freq = int(float(freq_m.group(1)) * 1_000_000) if freq_m else DEFAULT_FREQ
+    return rate, freq
+
+
+def _magnitude(path: Path) -> Optional[np.ndarray]:
+    raw = np.fromfile(path, dtype=np.uint8)
+    if raw.size < 4:
+        return None
+    raw = raw.astype(np.float32) - 127.5  # centre uint8 IQ
+    i, q = raw[0::2], raw[1::2]
+    n = min(i.size, q.size)
+    if n == 0:
+        return None
+    return np.sqrt(i[:n] * i[:n] + q[:n] * q[:n])
+
+
+def _runs_to_pulses(hyst: np.ndarray, us_per_sample: float) -> List[int]:
+    """Run-length encode a boolean high/low mask into signed µs durations."""
+    if hyst.size == 0:
+        return []
+    # Indices where the level changes.
+    change = np.flatnonzero(np.diff(hyst.view(np.int8))) + 1
+    bounds = np.concatenate(([0], change, [hyst.size]))
+    pulses: List[int] = []
+    for a, b in zip(bounds[:-1], bounds[1:]):
+        dur = (b - a) * us_per_sample
+        if dur < MIN_SHORT_US:
+            continue
+        sign = 1 if hyst[a] else -1
+        pulses.append(int(round(dur)) * sign)
+    return pulses
+
+
+def _trim_gaps(pulses: List[int]) -> List[int]:
+    """Drop leading/trailing low (gap) runs so the train starts on a pulse."""
+    start = 0
+    while start < len(pulses) and pulses[start] < 0:
+        start += 1
+    end = len(pulses)
+    while end > start and pulses[end - 1] < 0:
+        end -= 1
+    return pulses[start:end]
+
+
+def demod_file(path: Path) -> Tuple[Optional[List[int]], int]:
+    """Demodulate one .cu8 file to (pulse_train | None, frequency_hz)."""
+    path = Path(path)
+    rate, freq = parse_rate_freq(path.name)
+    us_per_sample = 1_000_000.0 / rate
+    mag = _magnitude(path)
+    if mag is None or mag.size == 0:
+        return None, freq
+
+    # Threshold between the noise floor and the signal peak. OOK captures have
+    # a large high/low separation, so a fixed fraction of the range is robust;
+    # hysteresis is unnecessary given the >15 dB SNR of these test files.
+    lo = float(np.percentile(mag, 50))
+    hi = float(np.percentile(mag, 99))
+    if hi - lo < 1.0:  # no amplitude modulation (FSK / dead air)
+        return None, freq
+    thr = lo + 0.3 * (hi - lo)
+    hyst = mag > thr
+
+    pulses = _trim_gaps(_runs_to_pulses(hyst, us_per_sample))
+    if len(pulses) > MAX_KEEP_PULSES:
+        pulses = pulses[:MAX_KEEP_PULSES]
+    if not looks_like_ook(pulses):
+        return None, freq
+    return pulses, freq
+
+
+def looks_like_ook(pulses: List[int]) -> bool:
+    """True when the train has enough OOK structure to be a real signal."""
+    if pulses is None or len(pulses) < MIN_TRANSITIONS:
+        return False
+    highs = [p for p in pulses if p > 0]
+    if len(highs) < MIN_TRANSITIONS // 2:
+        return False
+    # A single constant width with no variation is not a keyed message.
+    return len(set(highs)) > 1
+
+
+if __name__ == "__main__":
+    import argparse
+
+    ap = argparse.ArgumentParser(description="Demodulate an rtl_433 .cu8 to a pulse train")
+    ap.add_argument("file", type=Path)
+    args = ap.parse_args()
+    pulses, freq = demod_file(args.file)
+    if pulses is None:
+        print(f"{args.file.name}: no OOK pulse train (FSK/degenerate)")
+    else:
+        print(f"{args.file.name}: freq={freq} pulses={len(pulses)}")
+        print("first 24:", pulses[:24])

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()

scripts/train_cnn_classifier.py

@@ -0,0 +1,306 @@
+#!/usr/bin/env python3
+"""
+Device-Category 1D-CNN — Phase 3B (PLAN_TO_PROD)
+================================================
+
+Trains a 1D convolutional net that predicts a device *category* directly from
+the normalized RAW pulse array of a Sub-GHz capture — the "shape of the signal"
+rather than the hand-picked timing statistics the Phase 3A model uses.
+
+Architecture (per PLAN_TO_PROD §Phase B):
+    Conv1D(64,3) → BN → ReLU
+    Conv1D(128,3) → BN → ReLU
+    Conv1D(256,3) → BN → ReLU
+    GlobalAvgPool → Dense(256) → Dropout(0.3) → Dense(N) → Softmax
+Input: pulse array scaled to [-1,1], padded/truncated to 512 (pulse_encoder).
+
+Honesty guardrails (identical discipline to the Phase 3A trainer):
+    * GROUP-AWARE split — a device sub-folder never spans train/test, so we
+      don't score inflated accuracy off near-duplicate repeat captures.
+    * Epoch selection uses a validation slice carved from TRAIN only; the
+      group-disjoint TEST set is scored exactly once at the end.
+    * The heuristic router AND the Phase 3A statistical model are scored on the
+      *same* held-out files → a fair three-way comparison, not cherry-picked.
+    * Class-weighted loss + balanced accuracy reported, because the RAW corpus
+      is ~70% Garage/Gate and raw accuracy alone is misleading.
+
+Run:
+    python scripts/train_cnn_classifier.py
+    python scripts/train_cnn_classifier.py --per-category 400 --epochs 60
+"""
+
+import argparse
+import json
+import sys
+import time
+from collections import Counter
+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 src.matcher.pulse_encoder import encode_pulses, PULSE_LEN
+from scripts.benchmark_realworld import FOLDER_TO_CATEGORY, DEFAULT_DATASET
+from scripts.train_category_classifier import (
+    extract_features, heuristic_top1, FEATURE_NAMES,
+)
+
+
+def collect(dataset_root: Path, per_category: int, seed: int):
+    """Return aligned CNN arrays + stat features + labels/groups/freqs.
+
+    Every accepted RAW file yields BOTH the CNN encoding and the Phase-3A
+    14-feature vector, indexed identically, so the same group split scores all
+    three models on the same files.
+    """
+    import random
+    rng = random.Random(seed)
+    ta, pd = get_timing_analyzer(), get_preamble_detector()
+
+    Xcnn, Xstat, 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
+            enc = encode_pulses(meta.raw_data)
+            if enc is None:
+                skipped["bad_pulses"] += 1
+                continue
+            feats = extract_features(meta.raw_data, meta.frequency, ta, pd)
+            if feats is None:
+                skipped["no_timing"] += 1
+                continue
+            Xcnn.append(enc)
+            Xstat.append(feats)
+            y.append(category)
+            groups.append(str(p.parent))
+            freqs.append(meta.frequency)
+            taken += 1
+    return (np.asarray(Xcnn, dtype=np.float32),
+            np.asarray(Xstat, dtype=np.float32),
+            np.array(y), np.array(groups), np.array(freqs), skipped)
+
+
+def build_model(n_classes: int):
+    import torch.nn as nn
+
+    class PulseCNN(nn.Module):
+        def __init__(self, n_out):
+            super().__init__()
+            self.features = nn.Sequential(
+                nn.Conv1d(1, 64, 3, padding=1), nn.BatchNorm1d(64), nn.ReLU(),
+                nn.MaxPool1d(2),
+                nn.Conv1d(64, 128, 3, padding=1), nn.BatchNorm1d(128), nn.ReLU(),
+                nn.MaxPool1d(2),
+                nn.Conv1d(128, 256, 3, padding=1), nn.BatchNorm1d(256), nn.ReLU(),
+                nn.AdaptiveAvgPool1d(1),  # GlobalAvgPool
+            )
+            self.head = nn.Sequential(
+                nn.Flatten(),
+                nn.Linear(256, 256), nn.ReLU(), nn.Dropout(0.3),
+                nn.Linear(256, n_out),
+            )
+
+        def forward(self, x):
+            return self.head(self.features(x))
+
+    return PulseCNN(n_classes)
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--dataset", type=Path, default=DEFAULT_DATASET)
+    ap.add_argument("--per-category", type=int, default=400)
+    ap.add_argument("--seed", type=int, default=42)
+    ap.add_argument("--test-size", type=float, default=0.25)
+    ap.add_argument("--val-size", type=float, default=0.15)
+    ap.add_argument("--epochs", type=int, default=60)
+    ap.add_argument("--batch-size", type=int, default=64)
+    ap.add_argument("--lr", type=float, default=1e-3)
+    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)
+
+    import torch
+    import torch.nn as nn
+    from sklearn.model_selection import GroupShuffleSplit
+    from sklearn.metrics import (accuracy_score, balanced_accuracy_score,
+                                 f1_score, confusion_matrix, classification_report)
+
+    torch.manual_seed(args.seed)
+    np.random.seed(args.seed)
+
+    print("Collecting + encoding RAW pulses ...")
+    t0 = time.time()
+    Xcnn, Xstat, y, groups, freqs, skipped = collect(
+        args.dataset, args.per_category, args.seed)
+    print(f"  {len(Xcnn)} 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))}")
+
+    classes = sorted(set(y))
+    cls_to_i = {c: i for i, c in enumerate(classes)}
+    yi = np.array([cls_to_i[c] for c in y])
+
+    # Group-aware test split (no device leakage)
+    gss = GroupShuffleSplit(n_splits=1, test_size=args.test_size,
+                            random_state=args.seed)
+    trainval_idx, test_idx = next(gss.split(Xcnn, yi, groups))
+    # Validation carved from train (for epoch selection only; test untouched)
+    gss2 = GroupShuffleSplit(n_splits=1, test_size=args.val_size,
+                             random_state=args.seed)
+    tr_rel, val_rel = next(gss2.split(Xcnn[trainval_idx], yi[trainval_idx],
+                                      groups[trainval_idx]))
+    train_idx = trainval_idx[tr_rel]
+    val_idx = trainval_idx[val_rel]
+    print(f"  train={len(train_idx)} val={len(val_idx)} test={len(test_idx)} "
+          f"(test groups={len(set(groups[test_idx]))}, group-disjoint)")
+
+    def to_tensor(idx):
+        x = torch.from_numpy(Xcnn[idx]).unsqueeze(1)  # (N,1,L)
+        t = torch.from_numpy(yi[idx]).long()
+        return x, t
+
+    Xtr, ytr = to_tensor(train_idx)
+    Xval, yval = to_tensor(val_idx)
+    Xte, yte = to_tensor(test_idx)
+
+    # Class weights (inverse freq) for imbalance
+    counts = np.bincount(yi[train_idx], minlength=len(classes)).astype(float)
+    counts[counts == 0] = 1.0
+    weights = torch.tensor((counts.sum() / counts), dtype=torch.float32)
+    weights = weights / weights.mean()
+
+    model = build_model(len(classes))
+    opt = torch.optim.Adam(model.parameters(), lr=args.lr, weight_decay=1e-4)
+    lossf = nn.CrossEntropyLoss(weight=weights)
+
+    n = len(train_idx)
+    best_val, best_state = -1.0, None
+    for epoch in range(args.epochs):
+        model.train()
+        perm = torch.randperm(n)
+        for i in range(0, n, args.batch_size):
+            bi = perm[i:i + args.batch_size]
+            opt.zero_grad()
+            out = model(Xtr[bi])
+            loss = lossf(out, ytr[bi])
+            loss.backward()
+            opt.step()
+        # Validation balanced accuracy
+        model.eval()
+        with torch.no_grad():
+            vpred = model(Xval).argmax(1).numpy()
+        vbal = balanced_accuracy_score(yval.numpy(), vpred)
+        if vbal > best_val:
+            best_val = vbal
+            best_state = {k: v.clone() for k, v in model.state_dict().items()}
+        if (epoch + 1) % 10 == 0:
+            print(f"  epoch {epoch+1:3d}  val_bal_acc={vbal:.1%}  best={best_val:.1%}")
+
+    model.load_state_dict(best_state)
+    model.eval()
+    with torch.no_grad():
+        te_logits = model(Xte)
+        te_pred = te_logits.argmax(1).numpy()
+    yte_np = yte.numpy()
+    acc = accuracy_score(yte_np, te_pred)
+    bal = balanced_accuracy_score(yte_np, te_pred)
+    f1 = f1_score(yte_np, te_pred, average="macro")
+
+    # ── Baselines on the SAME test files ───────────────────────────────────
+    router = get_category_router()
+    h_top1 = 0
+    for row, freq, truth in zip(Xstat[test_idx], freqs[test_idx], y[test_idx]):
+        cat, _allowed, _full = heuristic_top1(row, freq, router)
+        if cat == truth:
+            h_top1 += 1
+    h_acc = h_top1 / len(test_idx)
+
+    stat_acc = stat_bal = None
+    try:
+        import joblib
+        bundle = joblib.load(args.out / "category_classifier.joblib")
+        stat_model, stat_classes = bundle["model"], [str(c) for c in bundle["classes"]]
+        stat_pred = stat_model.predict(Xstat[test_idx])
+        stat_acc = accuracy_score(y[test_idx], stat_pred)
+        stat_bal = balanced_accuracy_score(y[test_idx], stat_pred)
+    except Exception as e:
+        print(f"  (statistical baseline unavailable: {e})")
+
+    print("\n" + "=" * 66)
+    print("THREE-WAY COMPARISON (same group-disjoint test set)")
+    print("=" * 66)
+    print(f"{'model':26}{'top-1':>10}{'balanced':>11}")
+    print(f"{'heuristic router':26}{h_acc:>10.1%}{'—':>11}")
+    if stat_acc is not None:
+        print(f"{'statistical (RF/GB)':26}{stat_acc:>10.1%}{stat_bal:>11.1%}")
+    print(f"{'1D CNN (this run)':26}{acc:>10.1%}{bal:>11.1%}")
+    print(f"\nCNN macro F1: {f1:.1%}")
+    print("\nPer-category (CNN):")
+    print(classification_report(yte_np, te_pred,
+                                labels=list(range(len(classes))),
+                                target_names=classes, zero_division=0))
+    print("Confusion (rows=truth, cols=pred):  labels=", classes)
+    print(confusion_matrix(yte_np, te_pred, labels=list(range(len(classes)))))
+
+    # ── Persist: torch state + ONNX + metrics ──────────────────────────────
+    args.out.mkdir(parents=True, exist_ok=True)
+    torch.save({"state_dict": best_state, "classes": classes,
+                "pulse_len": PULSE_LEN}, args.out / "category_cnn.pt")
+    dummy = torch.zeros(1, 1, PULSE_LEN)
+    onnx_path = args.out / "category_cnn.onnx"
+    try:
+        torch.onnx.export(
+            model, dummy, str(onnx_path),
+            input_names=["pulses"], output_names=["logits"],
+            dynamic_axes={"pulses": {0: "batch"}, "logits": {0: "batch"}},
+            opset_version=13,
+        )
+        onnx_ok = True
+    except Exception as e:
+        onnx_ok = False
+        print(f"  (ONNX export failed, torch .pt still saved: {e})")
+    meta = {
+        "n_samples": int(len(Xcnn)), "n_train": int(len(train_idx)),
+        "n_val": int(len(val_idx)), "n_test": int(len(test_idx)),
+        "classes": classes, "pulse_len": PULSE_LEN, "epochs": args.epochs,
+        "class_balance": {k: int(v) for k, v in Counter(y).items()},
+        "cnn": {"accuracy": acc, "balanced_accuracy": bal, "macro_f1": f1},
+        "heuristic_same_test": {"top1": h_acc},
+        "statistical_same_test": (
+            {"top1": stat_acc, "balanced_accuracy": stat_bal}
+            if stat_acc is not None else None),
+    }
+    (args.out / "category_cnn_metrics.json").write_text(json.dumps(meta, indent=2))
+    print(f"\nSaved torch -> {args.out / 'category_cnn.pt'}")
+    if onnx_ok:
+        print(f"Saved ONNX  -> {onnx_path}")
+    print(f"Saved metrics-> {args.out / 'category_cnn_metrics.json'}")
+
+
+if __name__ == "__main__":
+    main()

src/api/main_simple.py

@@ -557,7 +557,11 @@ def _build_sub_capture(content: bytes, filename: str, entry: dict,
             {
                 "device_name": m.device_name,
                 "manufacturer": m.manufacturer,
-                "category": m.match_details.get("predicted_category"),
+                # Per-device category is that device's own catalog category
+                # (e.g. an Acurite sensor stays "Weather Sensor"). The overall
+                # ML call lives in the top-level `device_category` field.
+                "category": m.match_details.get("device_category")
+                or m.match_details.get("predicted_category"),
                 "confidence": m.confidence,
                 "method": m.match_method,
                 "details": m.match_details,

src/matcher/category_classifier.py

@@ -0,0 +1,154 @@
+#!/usr/bin/env python3
+"""
+Statistical Category Classifier — inference wrapper (PLAN_TO_PROD Phase 3A)
+==========================================================================
+
+Loads the supervised device-category model trained by
+``scripts/train_category_classifier.py`` and predicts a device *category*
+(with class probabilities) from the timing/statistical features of a RAW
+Sub-GHz capture.
+
+This is the "statistical" leg of the designed ensemble. It is used as a
+category *prior* over the heuristic RAW matches (see
+``pattern_decoder._apply_ml_category_boost``) — it re-ranks toward the
+category the model believes the signal belongs to, but never invents a match.
+
+Graceful degradation: if the model file is missing or joblib/sklearn are
+unavailable, ``available`` is False and the decoder falls back to the pure
+heuristic path unchanged.
+
+Feature parity: the vector is built in the frozen order carried in the model
+bundle (``features``), matching ``train_category_classifier.extract_features``
+exactly. If the two ever diverge, retrain rather than hand-edit one side.
+"""
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Dict, List, Optional
+
+import numpy as np
+
+from src.matcher.timing_analyzer import get_timing_analyzer
+from src.matcher.preamble_detector import get_preamble_detector
+
+DEFAULT_MODEL_PATH = (
+    Path(__file__).parent.parent.parent / "models" / "category_classifier.joblib"
+)
+
+# Must match scripts/train_category_classifier._PREAMBLE_ID
+_PREAMBLE_ID = {"none": 0, "long_burst": 1, "alternating": 2,
+                "sync_word": 3, "custom": 4}
+
+
+@dataclass
+class MLCategoryPrediction:
+    """Result of a statistical category prediction."""
+    available: bool = False
+    top_category: Optional[str] = None
+    top_prob: float = 0.0
+    probabilities: Dict[str, float] = field(default_factory=dict)
+
+
+class MLCategoryClassifier:
+    """Lazy-loading inference wrapper around the trained sklearn model."""
+
+    def __init__(self, model_path: Path = DEFAULT_MODEL_PATH):
+        self.model_path = Path(model_path)
+        self._loaded = False
+        self._model = None
+        self._features: List[str] = []
+        self._classes: List[str] = []
+        self.timing_analyzer = get_timing_analyzer()
+        self.preamble_detector = get_preamble_detector()
+
+    @property
+    def available(self) -> bool:
+        self._ensure_loaded()
+        return self._model is not None
+
+    def _ensure_loaded(self):
+        if self._loaded:
+            return
+        self._loaded = True
+        try:
+            import joblib
+            bundle = joblib.load(self.model_path)
+            self._model = bundle["model"]
+            self._features = list(bundle["features"])
+            self._classes = [str(c) for c in bundle["classes"]]
+        except Exception:
+            # Missing model, joblib, or sklearn — degrade to heuristic-only.
+            self._model = None
+
+    def _feature_vector(self, pulses, frequency) -> Optional[List[float]]:
+        """Build the frozen-order feature vector for one RAW capture.
+
+        Mirrors ``train_category_classifier.extract_features`` — keep in sync.
+        """
+        if not pulses:
+            return None
+        timing = self.timing_analyzer.extract_timing(pulses)
+        short, long = timing.short_pulse_us, timing.long_pulse_us
+        if short == 0 or long == 0:
+            return None
+
+        detected = self.preamble_detector.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)
+
+        named = {
+            "freq_mhz": (frequency or 0) / 1_000_000,
+            "short_pulse_us": short,
+            "long_pulse_us": long,
+            "pulse_ratio": timing.pulse_ratio,
+            "short_gap_us": timing.short_gap_us,
+            "long_gap_us": timing.long_gap_us,
+            "gap_ratio": gap_ratio,
+            "duty_cycle": timing.duty_cycle,
+            "pulse_count": len(pulses),
+            "pulse_mean_abs": float(abs_p.mean()),
+            "pulse_std_abs": float(abs_p.std()),
+            "pulse_min_abs": float(abs_p.min()),
+            "pulse_max_abs": float(abs_p.max()),
+            "preamble_type": _PREAMBLE_ID.get(ptype, 0),
+        }
+        return [named[name] for name in self._features]
+
+    def predict(self, pulses, frequency) -> MLCategoryPrediction:
+        """Predict a device category from RAW pulses. Never raises."""
+        self._ensure_loaded()
+        if self._model is None:
+            return MLCategoryPrediction(available=False)
+
+        try:
+            vec = self._feature_vector(pulses, frequency)
+            if vec is None:
+                return MLCategoryPrediction(available=False)
+            X = np.asarray([vec], dtype=float)
+            probs = self._model.predict_proba(X)[0]
+            prob_map = {self._classes[i]: float(probs[i])
+                        for i in range(len(self._classes))}
+            top_i = int(np.argmax(probs))
+            return MLCategoryPrediction(
+                available=True,
+                top_category=self._classes[top_i],
+                top_prob=float(probs[top_i]),
+                probabilities=prob_map,
+            )
+        except Exception:
+            return MLCategoryPrediction(available=False)
+
+
+# Singleton
+_classifier: Optional[MLCategoryClassifier] = None
+
+
+def get_category_classifier() -> MLCategoryClassifier:
+    """Get singleton statistical category classifier."""
+    global _classifier
+    if _classifier is None:
+        _classifier = MLCategoryClassifier()
+    return _classifier

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:
+        if not signal_data.has_raw_data and not signal_data.protocol:
-            # No raw data - cannot identify
+            # 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)
+        # Step 2: Statistical classification (needs RAW pulses; skip on KEY)
-        if use_statistical and heuristic_matches:
+        if use_statistical and heuristic_matches and signal_data.has_raw_data:
             statistical_matches = self._apply_statistical_scoring(
                 signal_data,
                 heuristic_matches

src/matcher/engine.py

@@ -87,7 +87,14 @@ class SignatureMatcher:
                     manufacturer=device_match.manufacturer or "Unknown",
                     confidence=device_match.confidence,
                     match_method=device_match.match_method,
-                    match_details=device_match.details
+                    # Preserve each device's own catalog category so the API can
+                    # surface it per-device (e.g. an Acurite sensor stays
+                    # "Weather Sensor"); the ML overall call remains in the
+                    # match_details' predicted_category.
+                    match_details={
+                        **device_match.details,
+                        "device_category": device_match.category,
+                    }
                 ))
 
             return match_results

src/matcher/pattern_decoder.py

@@ -25,6 +25,7 @@ from src.matcher.timing_analyzer import get_timing_analyzer
 from src.matcher.preamble_detector import get_preamble_detector
 from src.matcher.frequency_fingerprint import get_frequency_fingerprinter
 from src.matcher.category_router import get_category_router
+from src.matcher.category_classifier import get_category_classifier
 
 
 @dataclass
@@ -91,6 +92,7 @@ class PatternDecoder:
         self.preamble_detector = get_preamble_detector()
         self.frequency_fingerprinter = get_frequency_fingerprinter()
         self.category_router = get_category_router()
+        self.ml_classifier = get_category_classifier()
 
     def decode(self, metadata: SignalMetadata) -> List[DeviceMatch]:
         """
@@ -103,6 +105,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
@@ -123,7 +129,48 @@ class PatternDecoder:
         # Already integrated into timing_matches above
 
         # Deduplicate and rank by confidence
-        return self._rank_matches(matches)
+        matches = self._rank_matches(matches)
+
+        # Strategy 4 (Phase 3A): statistical category classifier. The trained
+        # model predicts a device *category* far better than the heuristic
+        # router on RAW captures, so when confident it owns the user-facing
+        # category, re-ranks candidates toward that family, and — when no
+        # protocol matched at all — still supplies a category-only result so
+        # the upload isn't left unidentified. No-op if the model is absent.
+        matches = self._apply_ml_category(matches, pulses, frequency)
+
+        return 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]:
         """
@@ -369,6 +416,91 @@ class PatternDecoder:
 
         return matches
 
+    def _apply_ml_category(
+        self,
+        matches: List[DeviceMatch],
+        pulses: List[int],
+        frequency: int,
+    ) -> List[DeviceMatch]:
+        """
+        Apply the statistical category classifier to the RAW decode result.
+
+        The classifier predicts a device *category* far more accurately than
+        the heuristic router on RAW captures (Phase 3A: ~62.5% balanced vs
+        ~30%), so when it is confident it:
+
+        1. Becomes the source of the user-facing ``predicted_category`` on every
+           match (the field the live upload path surfaces).
+        2. Re-ranks candidate protocols toward the believed family (``BOOST``,
+           0.25, mirrors the designed ensemble's statistical weight).
+        3. When *no* protocol matched, emits a single category-only match so
+           the capture still receives a device category instead of nothing —
+           the whole point of Phase 3A category-level ID for unknown devices.
+
+        Degrades to a pure no-op when the model is unavailable or too unsure
+        (``top_prob`` below ``MIN_PROB``), preserving the heuristic path exactly.
+        """
+        pred = self.ml_classifier.predict(pulses, frequency)
+        if not pred.available or not pred.top_category:
+            return matches
+
+        BOOST = 0.25            # statistical leg weight (ensemble design)
+        MIN_PROB = 0.40         # ignore low-confidence predictions
+
+        if pred.top_prob < MIN_PROB:
+            return matches
+
+        # No protocol matched: supply a category-only result (like KEY files).
+        if not matches:
+            signature = ProtocolSignature(
+                name=f"Unknown ({pred.top_category})",
+                category=pred.top_category,
+                frequency=frequency or 433920000,
+            )
+            # Category-level ID without protocol confirmation — cap the reported
+            # confidence so it never masquerades as a precise device match.
+            CATEGORY_ONLY_CEIL = 0.75
+            return [DeviceMatch(
+                protocol=signature,
+                confidence=round(min(pred.top_prob, CATEGORY_ONLY_CEIL), 3),
+                match_method="ml_category",
+                details={
+                    "predicted_category": pred.top_category,
+                    "category_source": "ml_statistical",
+                    "ml_category_prob": f"{pred.top_prob:.2%}",
+                    "source": "ml_category_only",
+                },
+            )]
+
+        adjusted = []
+        for m in matches:
+            agree_prob = pred.probabilities.get(m.protocol.category, 0.0)
+            # +top_prob if this protocol is in the predicted category; else a
+            # soft penalty scaled by how much less mass the model gave it.
+            if m.protocol.category == pred.top_category:
+                factor = 1.0 + BOOST * pred.top_prob
+            else:
+                factor = 1.0 - BOOST * (pred.top_prob - agree_prob)
+            new_conf = max(0.0, min(1.0, m.confidence * factor))
+            heuristic_category = m.details.get("predicted_category")
+            adjusted.append(DeviceMatch(
+                protocol=m.protocol,
+                confidence=new_conf,
+                match_method=m.match_method,
+                details={
+                    **m.details,
+                    # ML is the better RAW category predictor -> it owns the
+                    # user-facing category; keep the heuristic one for audit.
+                    "predicted_category": pred.top_category,
+                    "heuristic_category": heuristic_category,
+                    "category_source": "ml_statistical",
+                    "ml_category_prob": f"{pred.top_prob:.2%}",
+                    "ml_boost_applied": f"{factor:.3f}",
+                },
+            ))
+
+        return sorted(adjusted, key=lambda m: m.confidence, reverse=True)
+
     def _apply_spread_penalty(self, matches: List[DeviceMatch]) -> List[DeviceMatch]:
         """
         Reduce confidence when top matches are too close together.

src/matcher/pulse_encoder.py

@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+"""
+Shared RAW-pulse encoder for the 1D-CNN (Phase 3B).
+
+A single source of truth for turning a Flipper RAW pulse array into the fixed
+normalized tensor the CNN consumes — imported by BOTH the trainer
+(``scripts/train_cnn_classifier.py``) and the inference wrapper
+(``src/matcher/cnn_classifier.py``) so the two can never drift.
+
+Encoding:
+    * take the first ``length`` signed durations (µs; +high / -low),
+    * scale by the sequence's *median absolute* duration — a robust, TE-like
+      unit so a short pulse is ~±1 and a long pulse ~±2-3 (the ratios that carry
+      the protocol identity), while absolute gain / capture level is normalised
+      out. Using the median instead of the max is deliberate: a single huge
+      inter-packet gap (~10 000 µs) would otherwise squash every informative
+      short pulse toward zero and blind the CNN.
+    * clip to ±``CLIP`` and rescale to [-1, 1] so outsized gaps saturate the
+      ceiling instead of dominating the dynamic range,
+    * right-pad with zeros to ``length``.
+
+Returns a float32 array of shape ``(length,)`` or ``None`` when the pulse train
+is empty or degenerate (all-zero).
+"""
+
+from typing import List, Optional
+
+import numpy as np
+
+PULSE_LEN = 512
+CLIP = 8.0  # in median-pulse units; gaps beyond this saturate to ±1
+
+
+def encode_pulses(pulses: List[int], length: int = PULSE_LEN) -> Optional[np.ndarray]:
+    if pulses is None or len(pulses) == 0:
+        return None
+    arr = np.asarray(pulses[:length], dtype=np.float32)
+    nz = np.abs(arr[arr != 0])
+    if nz.size == 0:
+        return None
+    scale = float(np.median(nz))
+    if scale <= 0.0:
+        scale = float(np.max(np.abs(arr)))
+    if scale <= 0.0:
+        return None
+    arr = np.clip(arr / scale, -CLIP, CLIP) / CLIP
+    if arr.size < length:
+        arr = np.pad(arr, (0, length - arr.size))
+    return arr.astype(np.float32)

src/parser/sub_parser.py

@@ -182,6 +182,20 @@ class SubFileParser:
                 data_hex = self.fields['Data_RAW'].replace(' ', '')
                 metadata.bin_raw_data = bytes.fromhex(data_hex)
 
+            # Reconstruct a signed pulse train (µs) from the demodulated bit
+            # stream so BinRAW captures flow through the same RAW identification
+            # path as RAW_Data files. Data_RAW is the signal sampled at TE-µs
+            # steps: a run of N identical bits is one pulse of N*TE µs (+high on
+            # 1s, -low on 0s). This populates raw_data, which makes the
+            # has_raw_data property true.
+            n_bits = metadata.bin_raw_bit or metadata.bit_length
+            if metadata.bin_raw_data and metadata.bin_raw_te and n_bits:
+                pulses = self._binraw_to_pulses(
+                    metadata.bin_raw_data, n_bits, metadata.bin_raw_te)
+                if pulses:
+                    metadata.raw_data = pulses
+                    metadata.pulse_count = len(pulses)
+
             # Calculate duration
             if metadata.bit_length and metadata.bin_raw_te:
                 metadata.duration_ms = (metadata.bit_length * metadata.bin_raw_te) / 1000.0
@@ -190,6 +204,32 @@ class SubFileParser:
             self.errors.append(f"Error parsing BinRAW fields: {e}")
             logger.warning(f"Parse error: {e}")
 
+    @staticmethod
+    def _binraw_to_pulses(data: bytes, n_bits: int, te: int) -> List[int]:
+        """Expand a BinRAW bit stream into a signed timing array (µs).
+
+        Bits are read MSB-first, truncated to ``n_bits``, then run-length
+        encoded: each run of ``k`` equal bits becomes ``k*te`` µs, signed
+        positive for 1s (pulse) and negative for 0s (gap).
+        """
+        bits = []
+        for byte in data:
+            for i in range(7, -1, -1):
+                bits.append((byte >> i) & 1)
+        bits = bits[:n_bits]
+        if not bits:
+            return []
+        pulses: List[int] = []
+        current, run = bits[0], 1
+        for b in bits[1:]:
+            if b == current:
+                run += 1
+            else:
+                pulses.append(run * te * (1 if current else -1))
+                current, run = b, 1
+        pulses.append(run * te * (1 if current else -1))
+        return pulses
+
 
 def parse_sub_file(file_path: str) -> SignalMetadata:
     """