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

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

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

.dockerignore

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

Dockerfile

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

docker-compose.yml

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

models/category_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/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_cnn_classifier.py

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

src/api/main_simple.py

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

src/matcher/category_classifier.py

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

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

src/matcher/pulse_encoder.py

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

src/parser/sub_parser.py

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