Merge feat/decoded-key-exact-db-match: exact signature-DB match for decoded KEY .sub files

10 offline tests for the KEY-file identification path: a named protocol that
is in the DB resolves to the real catalog signature (category + manufacturer)
via match_method 'decoded_key_exact' with frequency-aware confidence
(0.95 consistent / 0.90 off-band); unknown names fall back to the router
('decoded_key', <0.90) yet stay identified; plus _lookup_known_protocol
case-insensitivity and None handling. Values are drawn from the DB itself so
the suite tracks catalog changes. Bare pytest, no server.

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

.gitignore

@@ -106,3 +106,10 @@ site/
 # External signature databases (git repositories)
 signatures/flipperzero-firmware/
 signatures/rtl_433/
+
+# UI verification screenshots (scripts/verify_ui.mjs)
+verify-shots/
+
+# Trained model binaries (regenerated by training scripts, not versioned)
+models/*.onnx
+models/*.pt

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"]

VERIFICATION.md

@@ -0,0 +1,304 @@
+# GigLez — Deployable-Artifact Verification
+
+**Purpose:** prove that the *thing you actually ship* — the Docker container running
+`src/api/main_simple.py` — serves the **full user journey**, end to end, the way a real
+user experiences it. Unit tests and in-process checks passing is **not** the same as the
+shipped app working. This doc is self-contained: you can clear context, come back, and run
+it top to bottom.
+
+Last grounded against the code on **2026-07-20** (branch `master`).
+
+---
+
+## 0. What "the deployable artifact" actually is
+
+| Layer | What it is | Where |
+|-------|-----------|-------|
+| Container | `python:3.10-slim`, non-root, torch-free (417 MB) | `Dockerfile` |
+| Process | `uvicorn src.api.main_simple:app` on `0.0.0.0:8000` | `docker-compose.yml` |
+| Served UI | One SPA page + 6 JS modules | `templates/index.html`, `static/js/{detail-modal,map,upload,search,stats,main}.js` |
+| API | FastAPI JSON endpoints (see §3) | `main_simple.py` |
+| Data store | JSON file `{captures:[...], upload_counter, last_updated}` on a **named volume** `giglez_data:/app/data` | `data/captures_simple.json` |
+| ML model | `models/category_classifier.joblib` (sklearn 1.6.1 / numpy 2.2.6 — pins must match) | baked into image |
+
+**Runtime external dependencies (important):** the UI pulls Leaflet, Leaflet.markercluster,
+Leaflet.heat, and Chart.js from **unpkg/jsdelivr CDNs**, and map tiles from the **OSM tile
+server**. The artifact is **not self-contained for offline use** — an air-gapped host will
+render a broken map. If offline deploy is ever a requirement, vendor these assets locally.
+
+---
+
+## 1. The full user journey (the 9 things to prove)
+
+```
+1. BUILD    docker compose up --build → container reports healthy
+2. LOAD     GET / returns the SPA; all 6 JS + CSS assets load; no console errors
+3. MAP      Leaflet map renders; existing captures show as markers/clusters; stat counters populate
+4. UPLOAD   user picks a .sub, sets GPS (or "use my location"), submits → progress → result card with a device match
+5. IDENTIFY the uploaded capture gets a device_name + device_category + confidence (parse→match→ML ran)
+6. PLOT     the new capture appears on the map at its GPS
+7. SEARCH   query by protocol / frequency / geo returns the capture
+8. STATS    counters + distributions reflect the new data
+9. PERSIST  restart the container → data survives (named volume did its job)
+```
+
+Verification is layered so a failure tells you *where* it broke:
+
+- **§2 Layer 0** — artifact builds & boots
+- **§3 Layer 1** — API contract (deterministic curl/jq) → proves the backend serves the journey
+- **§4 Layer 2** — browser/UI (Playwright) → proves the *actual UI a user touches* works
+- **§5 Layer 3** — persistence across restart
+- **§6 Layer 4** — regression guardrails (things fixed recently that must stay fixed)
+
+---
+
+## 2. Layer 0 — Build & boot the real artifact
+
+```bash
+cd ~/coding/security/giglez
+
+# Build + run exactly what ships
+docker compose up --build -d
+
+# Wait for healthy (compose healthcheck hits /health). Give it ~15s.
+docker compose ps            # STATUS should show "healthy"
+docker inspect --format '{{.State.Health.Status}}' $(docker compose ps -q giglez)
+```
+
+**PASS:** container status = `healthy`.
+**If it fails:** `docker compose logs giglez` — look for import errors (usually a pin
+mismatch: the joblib model needs sklearn 1.6.1 / numpy 2.2.6; `requirements-prod.txt`, not
+the stale root `requirements.txt`).
+
+Set a base URL for the rest of the doc:
+
+```bash
+BASE=http://localhost:8000
+```
+
+> **Caveat when running inside Claude Code's shell:** the `rtk` hook intercepts *piped*
+> `curl` and returns a fake compressed body like `{"ok":true,"top":[]}`. Always write the
+> response to a file first (`curl -s "$BASE/..." -o /tmp/r.json`) then inspect with
+> `jq . /tmp/r.json`. In a normal human terminal this caveat does not apply.
+
+---
+
+## 3. Layer 1 — API contract checks (deterministic)
+
+These prove the backend serves stages 2–8 without a browser. Copy-paste each block; the
+expected shape is shown.
+
+### 3.1 Health + API banner + SPA HTML
+
+```bash
+curl -s "$BASE/health" -o /tmp/h.json;      jq . /tmp/h.json
+# expect: {"status":"healthy","database":"not_connected","mode":"simple"}
+
+curl -s "$BASE/api" -o /tmp/api.json;        jq .status /tmp/api.json
+# expect: "operational"
+
+curl -s "$BASE/" -o /tmp/index.html
+grep -c 'id="map"' /tmp/index.html          # expect: 1  (SPA shell served)
+```
+
+### 3.2 Static assets the page actually loads (stage 2)
+
+```bash
+for f in detail-modal map upload search stats main; do
+  code=$(curl -s -o /dev/null -w '%{http_code}' "$BASE/static/js/$f.js")
+  echo "$f.js -> $code"      # every one must be 200
+done
+curl -s -o /dev/null -w 'main.css -> %{http_code}\n' "$BASE/static/css/main.css"
+```
+
+**PASS:** all `200`. A `404` here = broken page for real users.
+
+### 3.3 Baseline data snapshot (stages 3 & 8)
+
+```bash
+curl -s "$BASE/api/v1/stats/summary" -o /tmp/s0.json
+jq '{total_captures, unique_devices, data_sources}' /tmp/s0.json
+BASELINE=$(jq .total_captures /tmp/s0.json); echo "baseline captures = $BASELINE"
+
+curl -s "$BASE/api/v1/query/captures?limit=1000" -o /tmp/q0.json
+jq '.total' /tmp/q0.json                     # should equal $BASELINE
+```
+
+### 3.4 Upload → identify → persist (stages 4, 5, 9-write)
+
+A minimal, self-contained RAW `.sub` (no dependency on the big dataset, which is *excluded*
+from the image):
+
+```bash
+cat > /tmp/sample.sub <<'SUB'
+Filetype: Flipper SubGhz RAW File
+Version: 1
+Frequency: 433920000
+Preset: FuriHalSubGhzPresetOok650Async
+Protocol: RAW
+RAW_Data: 350 -350 350 -700 700 -350 350 -350 700 -700 350 -350 350 -700 700 -350 350 -700 350 -350 700 -350 350 -700 350 -350 350 -700 700 -700 350 -350
+SUB
+
+MANIFEST='{"captures":[{"filename":"sample.sub","latitude":47.6062,"longitude":-122.3321,"timestamp":"2026-07-20T12:00:00Z"}]}'
+
+curl -s -X POST "$BASE/api/v1/captures/upload" \
+  -F "files=@/tmp/sample.sub" \
+  -F "manifest=$MANIFEST" -o /tmp/up.json
+
+jq '{success, total_successful, name:.successful[0].device_name,
+     category:.successful[0].device_category,
+     conf:.successful[0].match_confidence,
+     id:.successful[0].id}' /tmp/up.json
+```
+
+**PASS:** `success=true`, `total_successful=1`, and `device_name` / `device_category` /
+`id` are non-null (parse → match → ML category all ran). Note the returned `id`.
+
+### 3.5 New capture is queryable & plottable (stages 6 & 8)
+
+```bash
+curl -s "$BASE/api/v1/stats/summary" -o /tmp/s1.json
+echo "after = $(jq .total_captures /tmp/s1.json)  (expect baseline+1 = $((BASELINE+1)))"
+
+# The capture carries GPS so the map can plot it:
+NEWID=$(jq '.successful[0].id' /tmp/up.json)
+curl -s "$BASE/api/v1/captures/$NEWID" -o /tmp/det.json
+jq '.capture | {id, latitude, longitude, device_category}' /tmp/det.json
+```
+
+**PASS:** count = baseline+1; the capture has numeric `latitude`/`longitude`.
+
+### 3.6 Export (bonus)
+
+```bash
+curl -s "$BASE/api/v1/export" -o /tmp/export.ndjson
+head -1 /tmp/export.ndjson | jq .            # each line is one capture record
+```
+
+---
+
+## 4. Layer 2 — Browser / UI checks (the real user journey)
+
+The API can be perfect while the UI is broken (bad JS, CDN blocked, map never inits). This
+layer drives a **real headless browser** through the DOM. Node Playwright 1.58.2 is present.
+
+Run the bundled script:
+
+```bash
+cd ~/coding/security/giglez
+BASE=http://localhost:8000 node scripts/verify_ui.mjs
+```
+
+It performs, against real element IDs from `index.html`:
+
+1. `goto /`, collect **console errors** and **failed network requests**.
+2. Assert the Leaflet map container `#map` exists and Leaflet initialised (`.leaflet-container`).
+3. Read the `#total-captures` / `#unique-devices` counters.
+4. Switch to the Upload section (`a[href="#upload"]`), set `#file-input` to `/tmp/sample.sub`,
+   fill `#default-lat` / `#default-lon`, click the **Upload** button (`onclick="uploadFiles()"`).
+5. Wait for `#upload-results` to become visible and assert it shows a device match.
+6. Re-query `/api/v1/query/captures` and assert the total incremented.
+7. Save screenshots to `./verify-shots/` at each step.
+
+**PASS criteria (the script prints a final PASS/FAIL and exits non-zero on failure):**
+- 0 console errors, 0 failed asset requests
+- `.leaflet-container` present (map actually rendered)
+- upload results panel shows a device name + category
+- capture total incremented by 1
+
+If you'd rather check by eye: open `http://localhost:8000` in a browser, open DevTools
+Console (should be clean), confirm the map tiles draw, then walk Upload → Search yourself.
+
+---
+
+## 5. Layer 3 — Persistence across restart (stage 9)
+
+The whole point of the named volume: a redeploy/restart must not lose user data.
+
+```bash
+BEFORE=$(curl -s "$BASE/api/v1/stats/summary" | jq .total_captures 2>/dev/null); echo "before=$BEFORE"
+docker compose restart giglez
+sleep 12
+AFTER=$(curl -s "$BASE/api/v1/stats/summary" -o /tmp/sa.json && jq .total_captures /tmp/sa.json)
+echo "after=$AFTER"     # must equal before
+```
+
+**PASS:** `after == before`. If it resets, the volume mount or `save_captures()` path is
+wrong and every restart wipes contributions.
+
+> Harder test (proves the volume, not just the file): `docker compose down` (keeps named
+> volumes) then `docker compose up -d` and re-check the count. Use `down -v` only when you
+> intend to wipe the store.
+
+---
+
+## 6. Layer 4 — Regression guardrails (recently fixed — must stay fixed)
+
+### 6.1 BinRAW captures produce a category (commit `4396c74`)
+BinRAW `.sub` used to yield zero identification. Upload one and confirm a category comes back:
+
+```bash
+cat > /tmp/binraw.sub <<'SUB'
+Filetype: Flipper SubGhz RAW File
+Version: 1
+Frequency: 433920000
+Preset: FuriHalSubGhzPreset2FSKDev476Async
+Protocol: BinRAW
+Bit: 48
+TE: 400
+Bit_RAW: 48
+Data_RAW: A9 3C 5F 12 87 D4
+SUB
+curl -s -X POST "$BASE/api/v1/captures/upload" \
+  -F "files=@/tmp/binraw.sub" \
+  -F 'manifest={"captures":[{"filename":"binraw.sub","latitude":47.61,"longitude":-122.33,"timestamp":"2026-07-20T12:05:00Z"}]}' \
+  -o /tmp/b.json
+jq '.successful[0] | {device_category, match_confidence}' /tmp/b.json
+```
+**PASS:** `device_category` is non-null.
+
+### 6.2 Per-device category shows the device's OWN catalog category (commit `41850b5`)
+A named device must not be mislabeled with the ML overall call (e.g. "Acurite … → Fan
+Controller"). Inspect the match list from §3.4's upload:
+
+```bash
+jq '.successful[0] | {headline_category: .device_category,
+     per_device: [.matched_devices[] | {device_name, category}]}' /tmp/up.json
+```
+**PASS:** each `matched_devices[].category` is that device's real catalog category; the
+top-level `.device_category` may differ (it's the ML overall call) — that's expected.
+
+---
+
+## 7. Pass/fail scorecard
+
+| # | Stage | Check | Pass condition |
+|---|-------|-------|----------------|
+| 0 | Build | §2 | container `healthy` |
+| 2 | Load | §3.1–3.2 | SPA HTML + all 6 JS + CSS = 200 |
+| 3 | Map data | §3.3 | stats + query return baseline |
+| 4/5 | Upload+ID | §3.4 | success, device_name+category+id non-null |
+| 6/8 | Plot+Stats | §3.5 | count = baseline+1, GPS present |
+| 2–8 | **UI** | §4 | Playwright PASS, 0 console errors, map rendered, upload result shown |
+| 9 | Persist | §5 | count unchanged after restart |
+| R1 | BinRAW | §6.1 | category non-null |
+| R2 | Category | §6.2 | per-device catalog category correct |
+
+All green ⇒ the deployable artifact serves the full user journey and is safe to put in
+front of a real user (beta). Remaining true prod gaps are **operational**, not functional:
+public host, TLS/reverse proxy, and (optional) vendoring the CDN assets for offline use.
+
+---
+
+## 8. Teardown
+
+```bash
+docker compose down            # keep the named volume (data persists)
+# docker compose down -v       # ONLY if you want to wipe the capture store
+rm -f /tmp/sample.sub /tmp/binraw.sub /tmp/*.json /tmp/index.html /tmp/export.ndjson
+```
+
+**Do not** commit any captures created during verification — the JSON store is baked into
+the image as seed data. If you uploaded against a local (non-Docker) `main_simple`, strip
+your test records from `data/captures_simple.json` before committing (filter by the test
+GPS you used).

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

scripts/verify_ui.mjs

@@ -0,0 +1,105 @@
+#!/usr/bin/env node
+/**
+ * GigLez UI verification — drives a real headless browser through the full
+ * user journey against the deployed SPA, asserting the things the API layer
+ * cannot: the page loads clean, the Leaflet map renders, and an upload done
+ * *through the actual DOM* returns a device match and increments the store.
+ *
+ * Usage:
+ *   BASE=http://localhost:8000 node scripts/verify_ui.mjs
+ *
+ * Requires Node Playwright (present: 1.58.x). Exits non-zero on any failure.
+ * Screenshots are written to ./verify-shots/.
+ */
+import { chromium } from 'playwright';
+import { writeFileSync, mkdirSync } from 'node:fs';
+
+const BASE = process.env.BASE || 'http://localhost:8000';
+const SHOTS = 'verify-shots';
+mkdirSync(SHOTS, { recursive: true });
+
+const fails = [];
+const ok = (cond, msg) => { console.log(`${cond ? 'PASS' : 'FAIL'}  ${msg}`); if (!cond) fails.push(msg); };
+
+// A minimal self-contained RAW .sub (no dependency on the big dataset).
+const SAMPLE_SUB = [
+  'Filetype: Flipper SubGhz RAW File',
+  'Version: 1',
+  'Frequency: 433920000',
+  'Preset: FuriHalSubGhzPresetOok650Async',
+  'Protocol: RAW',
+  'RAW_Data: 350 -350 350 -700 700 -350 350 -350 700 -700 350 -350 350 -700 700 -350 350 -700 350 -350 700 -350 350 -700 350 -350 350 -700 700 -700 350 -350',
+].join('\n');
+
+const consoleErrors = [];
+const failedRequests = [];
+
+const browser = await chromium.launch();
+const ctx = await browser.newContext();
+const page = await ctx.newPage();
+
+page.on('console', (m) => { if (m.type() === 'error') consoleErrors.push(m.text()); });
+page.on('requestfailed', (r) => failedRequests.push(`${r.url()} (${r.failure()?.errorText})`));
+
+try {
+  // ---- Stage 2: load ----
+  const resp = await page.goto(BASE, { waitUntil: 'networkidle', timeout: 30000 });
+  ok(resp && resp.ok(), `GET / returned ${resp && resp.status()}`);
+  await page.screenshot({ path: `${SHOTS}/1-loaded.png`, fullPage: true });
+
+  // ---- Stage 3: map rendered ----
+  await page.waitForSelector('#map', { timeout: 10000 });
+  const leafletUp = await page.locator('#map .leaflet-container').count();
+  ok(leafletUp > 0, 'Leaflet map initialised (.leaflet-container present)');
+
+  const baselineTxt = await page.locator('#total-captures').innerText().catch(() => '0');
+  const baseline = parseInt(baselineTxt, 10) || 0;
+  console.log(`      baseline #total-captures = ${baseline}`);
+
+  // ---- Stage 4: switch to Upload, fill the real form ----
+  await page.click('a[href="#upload"]').catch(() => {});
+  await page.waitForSelector('#upload-section.section.active, #upload-section', { timeout: 5000 });
+
+  await page.setInputFiles('#file-input', {
+    name: 'sample.sub',
+    mimeType: 'application/octet-stream',
+    buffer: Buffer.from(SAMPLE_SUB),
+  });
+  await page.fill('#default-lat', '47.6062');
+  await page.fill('#default-lon', '-122.3321');
+  await page.screenshot({ path: `${SHOTS}/2-upload-ready.png`, fullPage: true });
+
+  // The upload button is a global onclick="uploadFiles()".
+  await page.click('button[onclick="uploadFiles()"]');
+
+  // ---- Stage 5: results panel shows a match ----
+  await page.waitForSelector('#upload-results', { state: 'visible', timeout: 20000 });
+  const resultsTxt = (await page.locator('#upload-results').innerText()).trim();
+  ok(resultsTxt.length > 0, 'upload results panel is populated');
+  ok(/success|processed|device|match|categ/i.test(resultsTxt), 'results mention a processed/device match');
+  await page.screenshot({ path: `${SHOTS}/3-results.png`, fullPage: true });
+
+  // ---- Stage 6/8: store incremented (checked via the API the UI uses) ----
+  const total = await page.evaluate(async () => {
+    const r = await fetch('/api/v1/query/captures?limit=1000');
+    return (await r.json()).total;
+  });
+  ok(total >= baseline + 1, `capture total incremented (${baseline} -> ${total})`);
+
+  // ---- clean-page assertions ----
+  ok(consoleErrors.length === 0, `no console errors (saw ${consoleErrors.length})`);
+  if (consoleErrors.length) consoleErrors.slice(0, 5).forEach((e) => console.log('        console:', e));
+  // CDN/tile requests may legitimately fail if offline — report but treat local assets strictly.
+  const localFailed = failedRequests.filter((u) => u.includes('/static/') || u.includes(BASE));
+  ok(localFailed.length === 0, `no failed LOCAL asset requests (saw ${localFailed.length})`);
+  if (failedRequests.length) failedRequests.slice(0, 8).forEach((u) => console.log('        reqfail:', u));
+} catch (err) {
+  ok(false, `unhandled error: ${err.message}`);
+  await page.screenshot({ path: `${SHOTS}/error.png`, fullPage: true }).catch(() => {});
+} finally {
+  await browser.close();
+}
+
+writeFileSync(`${SHOTS}/summary.json`, JSON.stringify({ fails, consoleErrors, failedRequests }, null, 2));
+console.log('\n' + (fails.length ? `FAILED (${fails.length}): ${fails.join(' | ')}` : 'ALL CHECKS PASSED'));
+process.exit(fails.length ? 1 : 0);

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:
-            # 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/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,103 @@ 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. Two cases:
+
+        1. The named protocol is in our signature database. This is the
+           strongest possible identification — the capture device already
+           demodulated and *named* the protocol, and we recognise it — so we
+           surface the REAL catalog signature (true manufacturer + category)
+           with high, frequency-aware confidence. This is an exact "database
+           match", not a timing guess, and it keeps the surfaced category
+           consistent with the catalog (e.g. Princeton -> Garage Door Opener)
+           instead of the router's coarser guess.
+
+        2. The name is unknown to us. Fall back to category routing so the
+           upload is still identified by family rather than silently dropped.
+        """
+        known = self._lookup_known_protocol(metadata.protocol, metadata.frequency)
+        if known is not None:
+            freq_ok = bool(metadata.frequency) and known.matches_frequency(metadata.frequency)
+            # Exact, already-decoded match against a known protocol. Confidence
+            # is high but never a literal 1.0 (the key payload itself is not
+            # cryptographically verified); a consistent frequency lifts it.
+            confidence = 0.95 if freq_ok else 0.90
+            details = {
+                "predicted_category": known.category,
+                "source": "decoded_key_exact",
+                "bit_length": metadata.bit_length,
+                "frequency_match": freq_ok,
+                "manufacturer": known.manufacturer,
+            }
+            return [DeviceMatch(
+                protocol=known,
+                confidence=confidence,
+                match_method="decoded_key_exact",
+                details=details,
+            )]
+
+        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 _lookup_known_protocol(
+        self, name: Optional[str], frequency: Optional[int]
+    ) -> Optional[ProtocolSignature]:
+        """Exact (case-insensitive) protocol-name lookup in the signature DB.
+
+        When several catalog entries share a name, prefer one whose frequency
+        band matches the capture; otherwise return the first. Returns None if
+        the name is unknown.
+        """
+        if not name:
+            return None
+        target = name.strip().lower()
+        candidates = [
+            p for p in self.protocol_db.get_all()
+            if p.name.strip().lower() == target
+        ]
+        if not candidates:
+            return None
+        if frequency:
+            for cand in candidates:
+                if cand.matches_frequency(frequency):
+                    return cand
+        return candidates[0]
 
     def _identify_pulse_widths(self, pulses: List[int]) -> Tuple[int, int, int, int]:
         """
@@ -369,6 +471,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:
     """

tests/unit/test_binraw_identification.py

@@ -0,0 +1,188 @@
+#!/usr/bin/env python3
+"""
+Offline unit tests for the parser's device-identification path.
+
+Covers two things the offline (no-server) pipeline is responsible for:
+
+  1. BinRAW -> RAW identification route (commit 4396c74): a BinRAW capture has
+     no RAW_Data array, so the parser must reconstruct a signed pulse train from
+     the demodulated Data_RAW bit stream. That reconstruction is what lets a
+     BinRAW file flow through the SAME identification path as a RAW file
+     (``has_raw_data`` becomes true), yielding device matches.
+
+  2. Per-match catalog-category surfacing (commit 41850b5): every device match
+     the engine returns must carry its OWN catalog category in
+     ``match_details['device_category']`` (e.g. an Acurite sensor stays a
+     "Weather Sensor"), not just the single ML overall call.
+
+Everything here is author-time / offline: bare ``pytest`` runs it with no
+Docker, no bound server, and no network. ``.sub`` fixtures are written to the
+per-test ``tmp_path`` because ``*.sub`` is git-ignored repo-wide, so embedding
+the content keeps the suite green on a fresh clone.
+"""
+
+import pytest
+
+from src.parser.sub_parser import SubFileParser
+from src.matcher.engine import SignatureMatcher
+from src.matcher.pattern_decoder import PatternDecoder
+from src.matcher.protocol_database import get_protocol_database
+
+
+# --- minimal, self-contained fixtures (written to tmp_path at runtime) -------
+
+# BinRAW: alternating bit stream (0xAA == 0b10101010) x 6 bytes = 48 bits.
+# Alternating bits mean every run has length 1, so the reconstructed pulse
+# train is exactly 48 pulses, each of magnitude TE (495 us), sign-alternating.
+BINRAW_SUB = "\n".join([
+    "Filetype: Flipper SubGhz Key File",
+    "Version: 1",
+    "Frequency: 868350000",
+    "Preset: FuriHalSubGhzPresetOok270Async",
+    "Protocol: BinRAW",
+    "Bit: 48",
+    "TE: 495",
+    "Bit_RAW: 48",
+    "Data_RAW: AA AA AA AA AA AA",
+    "",
+])
+
+# RAW: a small but real timing array (315 MHz OOK).
+RAW_SUB = "\n".join([
+    "Filetype: Flipper SubGhz RAW File",
+    "Version: 1",
+    "Frequency: 315000000",
+    "Preset: FuriHalSubGhzPresetOok650Async",
+    "Protocol: RAW",
+    "RAW_Data: 2980 -240 520 -980 520 -980 1000 -500 480 -1020 520 -980 1000 -500",
+    "",
+])
+
+
+def _write(tmp_path, name, content):
+    p = tmp_path / name
+    p.write_text(content)
+    return str(p)
+
+
+def _catalog_categories():
+    """Authoritative {protocol_name: catalog_category} from the protocol DB."""
+    return {proto.name: proto.category for proto in get_protocol_database().get_all()}
+
+
+def _valid_categories():
+    """The set of every catalog category the protocol DB knows about."""
+    return {proto.category for proto in get_protocol_database().get_all()}
+
+
+# --- BinRAW parsing ----------------------------------------------------------
+
+class TestBinRawParsing:
+    def test_binraw_fields_extracted(self, tmp_path):
+        meta = SubFileParser().parse(_write(tmp_path, "b.sub", BINRAW_SUB))
+        assert meta.file_format == "BinRAW"
+        assert meta.frequency == 868350000
+        assert meta.modulation == "OOK"
+        assert meta.bin_raw_te == 495
+        assert meta.bin_raw_bit == 48
+        assert meta.bin_raw_data == bytes.fromhex("AAAAAAAAAAAA")
+
+    def test_binraw_reconstructs_raw_pulse_train(self, tmp_path):
+        """The BinRAW->RAW route: Data_RAW must expand into raw_data pulses."""
+        meta = SubFileParser().parse(_write(tmp_path, "b.sub", BINRAW_SUB))
+        assert meta.has_raw_data is True
+        # 48 alternating bits -> 48 unit-length runs -> 48 pulses.
+        assert meta.pulse_count == 48
+        assert len(meta.raw_data) == 48
+        # Every pulse is exactly one TE, alternating sign.
+        assert all(abs(t) == 495 for t in meta.raw_data)
+        assert meta.raw_data[0] > 0 and meta.raw_data[1] < 0
+
+
+class TestBinRawToPulses:
+    """The run-length expansion is deterministic; test it directly."""
+
+    def test_run_length_encoding(self):
+        # 0xF0 == 11110000 -> run of 4 ones then 4 zeros.
+        pulses = SubFileParser._binraw_to_pulses(bytes([0xF0]), 8, 100)
+        assert pulses == [400, -400]
+
+    def test_alternating_bits(self):
+        # 0xAA == 10101010 -> 8 unit runs, alternating sign starting high.
+        pulses = SubFileParser._binraw_to_pulses(bytes([0xAA]), 8, 100)
+        assert pulses == [100, -100, 100, -100, 100, -100, 100, -100]
+
+    def test_truncates_to_n_bits(self):
+        # Only the first 4 bits (1111) of 0xF0 are used.
+        pulses = SubFileParser._binraw_to_pulses(bytes([0xF0]), 4, 50)
+        assert pulses == [200]
+
+
+# --- RAW parsing -------------------------------------------------------------
+
+class TestRawParsing:
+    def test_raw_fields_extracted(self, tmp_path):
+        meta = SubFileParser().parse(_write(tmp_path, "r.sub", RAW_SUB))
+        assert meta.file_format == "RAW"
+        assert meta.frequency == 315000000
+        assert meta.modulation == "OOK"
+        assert meta.has_raw_data is True
+        assert meta.pulse_count == 14
+        assert meta.raw_data[0] == 2980
+        assert meta.avg_pulse_width > 0
+
+
+# --- identification: BinRAW flows through RAW, categories surface -------------
+
+class TestBinRawIdentificationRoute:
+    def test_binraw_yields_matches_via_raw_path(self, tmp_path):
+        """A BinRAW capture identifies at all only because of the RAW route."""
+        meta = SubFileParser().parse(_write(tmp_path, "b.sub", BINRAW_SUB))
+        matches = SignatureMatcher().match(meta, max_results=5)
+        assert len(matches) > 0
+        for m in matches:
+            assert m.device_name  # non-empty device name
+            cat = m.match_details.get("device_category")
+            assert isinstance(cat, str) and cat, \
+                "every match must surface a per-device catalog category"
+
+    def test_binraw_category_is_valid_catalog_category(self, tmp_path):
+        """Every surfaced device_category is a real category from the catalog."""
+        meta = SubFileParser().parse(_write(tmp_path, "b.sub", BINRAW_SUB))
+        valid = _valid_categories()
+        matches = SignatureMatcher().match(meta, max_results=5)
+        assert len(matches) > 0
+        for m in matches:
+            assert m.match_details.get("device_category") in valid
+
+    def test_binraw_decoder_category_is_protocol_catalog_category(self, tmp_path):
+        """At the source, DeviceMatch.category (surfaced by the engine) == protocol.category."""
+        meta = SubFileParser().parse(_write(tmp_path, "b.sub", BINRAW_SUB))
+        decoded = PatternDecoder().decode(meta)
+        assert len(decoded) > 0
+        for dm in decoded:
+            assert dm.category == dm.protocol.category
+
+
+class TestRawIdentificationCategory:
+    def test_raw_match_surfaces_catalog_category(self, tmp_path):
+        meta = SubFileParser().parse(_write(tmp_path, "r.sub", RAW_SUB))
+        catalog = _catalog_categories()
+        matches = SignatureMatcher().match(meta, max_results=5)
+        assert len(matches) > 0
+        checked = 0
+        for m in matches:
+            cat = m.match_details.get("device_category")
+            assert isinstance(cat, str) and cat
+            if m.device_name in catalog:
+                assert cat == catalog[m.device_name]
+                checked += 1
+        assert checked > 0
+
+    def test_decoder_category_is_protocol_catalog_category(self, tmp_path):
+        """DeviceMatch.category (the source of the surfaced value) == protocol.category."""
+        meta = SubFileParser().parse(_write(tmp_path, "r.sub", RAW_SUB))
+        decoded = PatternDecoder().decode(meta)
+        assert len(decoded) > 0
+        for dm in decoded:
+            assert dm.category == dm.protocol.category

tests/unit/test_decoded_key_identification.py

@@ -0,0 +1,131 @@
+#!/usr/bin/env python3
+"""
+Offline unit tests for decoded (KEY) .sub identification against the signature DB.
+
+A Flipper KEY file carries no RAW pulses but *names* a decoded protocol (e.g.
+``Protocol: Princeton``). When that name is in our signature database this is the
+strongest possible identification -- the capture device already demodulated and
+named the protocol and we recognise it -- so it must:
+
+  - resolve to the REAL catalog signature (true manufacturer + category), not the
+    category router's coarser guess (Princeton is a Garage Door Opener in the
+    catalog, which the router previously mislabelled "Remote Control"), and
+  - score with high, frequency-aware confidence (exact database match), not the
+    ~0.45 fuzzy-guess confidence it used to get.
+
+Unknown protocol names must still fall back to category routing so an upload is
+identified by family rather than silently dropped ("accept any capture ->
+identify the device").
+
+Offline / author-time: bare ``pytest`` runs this with no docker, no server, no
+network. KEY .sub content is embedded and written to ``tmp_path``.
+"""
+
+import pytest
+
+from src.parser.sub_parser import SubFileParser
+from src.matcher.engine import SignatureMatcher
+from src.matcher.pattern_decoder import PatternDecoder
+from src.matcher.protocol_database import get_protocol_database
+
+
+def _key_sub(protocol: str, frequency: int, bit: int = 24, key: str = "00 00 00 00 00 95 D5 D4") -> str:
+    return "\n".join([
+        "Filetype: Flipper SubGhz Key File",
+        "Version: 1",
+        f"Frequency: {frequency}",
+        "Preset: FuriHalSubGhzPresetOok270Async",
+        f"Protocol: {protocol}",
+        f"Bit: {bit}",
+        f"Key: {key}",
+        "TE: 400",
+        "",
+    ])
+
+
+def _write(tmp_path, content, name="k.sub"):
+    p = tmp_path / name
+    p.write_text(content)
+    return str(p)
+
+
+def _identify(tmp_path, content):
+    meta = SubFileParser().parse(_write(tmp_path, content))
+    return meta, SignatureMatcher().match(meta, max_results=5)
+
+
+# A known catalog protocol and a known one that carries a manufacturer, both
+# taken from the DB itself so the tests stay correct if the catalog changes.
+_DB = get_protocol_database()
+_KNOWN = next(p for p in _DB.get_all())
+_KNOWN_WITH_MFR = next((p for p in _DB.get_all() if p.manufacturer), None)
+_UNKNOWN_NAME = "ZZ_NOT_A_REAL_PROTOCOL_9999"
+
+
+class TestKnownProtocolExactMatch:
+    def test_named_protocol_resolves_to_catalog_signature(self, tmp_path):
+        meta, matches = _identify(tmp_path, _key_sub(_KNOWN.name, _KNOWN.frequency))
+        assert meta.file_format == "KEY"
+        assert meta.is_decoded is True
+        assert len(matches) >= 1
+        top = matches[0]
+        assert top.device_name == _KNOWN.name
+        assert top.match_method == "decoded_key_exact"
+        # Category comes from the catalog, not the router's guess.
+        assert top.match_details.get("device_category") == _KNOWN.category
+
+    def test_exact_match_scores_high_when_frequency_consistent(self, tmp_path):
+        _, matches = _identify(tmp_path, _key_sub(_KNOWN.name, _KNOWN.frequency))
+        assert matches[0].confidence == pytest.approx(0.95)
+        assert matches[0].match_details.get("frequency_match") is True
+
+    def test_exact_match_still_high_but_lower_on_frequency_mismatch(self, tmp_path):
+        # Same known name, but a frequency far outside its band.
+        off_band = _KNOWN.frequency + 50_000_000
+        _, matches = _identify(tmp_path, _key_sub(_KNOWN.name, off_band))
+        top = matches[0]
+        assert top.match_method == "decoded_key_exact"
+        assert top.confidence == pytest.approx(0.90)
+        assert top.match_details.get("frequency_match") is False
+
+    @pytest.mark.skipif(_KNOWN_WITH_MFR is None, reason="no catalog protocol carries a manufacturer")
+    def test_real_manufacturer_is_surfaced(self, tmp_path):
+        _, matches = _identify(
+            tmp_path, _key_sub(_KNOWN_WITH_MFR.name, _KNOWN_WITH_MFR.frequency, bit=66)
+        )
+        assert matches[0].manufacturer == _KNOWN_WITH_MFR.manufacturer
+
+
+class TestUnknownProtocolFallback:
+    def test_unknown_name_falls_back_to_router(self, tmp_path):
+        _, matches = _identify(tmp_path, _key_sub(_UNKNOWN_NAME, 433920000))
+        assert len(matches) >= 1, "unknown decoded protocol must still be identified by family"
+        top = matches[0]
+        assert top.match_method == "decoded_key"
+        # A routed guess must not masquerade as an exact database match.
+        assert top.confidence < 0.90
+
+    def test_unknown_name_still_names_the_protocol(self, tmp_path):
+        _, matches = _identify(tmp_path, _key_sub(_UNKNOWN_NAME, 433920000))
+        assert matches[0].device_name == _UNKNOWN_NAME
+
+
+class TestLookupHelper:
+    def test_known_name_case_insensitive(self):
+        dec = PatternDecoder()
+        sig = dec._lookup_known_protocol(_KNOWN.name.upper(), _KNOWN.frequency)
+        assert sig is not None
+        assert sig.name == _KNOWN.name
+
+    def test_unknown_name_returns_none(self):
+        assert PatternDecoder()._lookup_known_protocol(_UNKNOWN_NAME, 433920000) is None
+
+    def test_empty_name_returns_none(self):
+        assert PatternDecoder()._lookup_known_protocol(None, 433920000) is None
+
+    def test_prefers_frequency_matching_candidate(self):
+        """If duplicate-named entries exist, the frequency-consistent one wins."""
+        dec = PatternDecoder()
+        sig = dec._lookup_known_protocol(_KNOWN.name, _KNOWN.frequency)
+        assert sig is not None
+        assert sig.matches_frequency(_KNOWN.frequency)