diff --git a/.github/workflows/csi-encoder-training.yml b/.github/workflows/csi-encoder-training.yml new file mode 100644 index 000000000..d8ac617fd --- /dev/null +++ b/.github/workflows/csi-encoder-training.yml @@ -0,0 +1,36 @@ +name: CSI encoder training harness + +on: + pull_request: + paths: + - 'scripts/csi_training/**' + - 'scripts/train-csi-encoder.py' + - 'scripts/requirements-csi-training.txt' + - 'scripts/tests/test_csi_training*.py' + - '.github/workflows/csi-encoder-training.yml' + workflow_dispatch: + +permissions: + contents: read + +jobs: + synthetic-cpu-contract: + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + with: + persist-credentials: false + - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 + with: + python-version: '3.12' + - name: Install pinned CPU training dependencies + run: | + python -m pip install torch==2.11.0 --index-url https://download.pytorch.org/whl/cpu + python -m pip install -r scripts/requirements-csi-training.txt + - name: Validate real training code using temporary synthetic data + env: + PYTHONPATH: scripts + run: python -m unittest discover -s scripts/tests -p 'test_csi_training*.py' -v + - name: Enforce capture data policy + run: bash scripts/csi-data-policy-check.sh --tracked diff --git a/docs/adr/ADR-419-reproducible-csi-encoder-training.md b/docs/adr/ADR-419-reproducible-csi-encoder-training.md new file mode 100644 index 000000000..28aef46a7 --- /dev/null +++ b/docs/adr/ADR-419-reproducible-csi-encoder-training.md @@ -0,0 +1,76 @@ +# ADR-419: Reproducible CSI encoder training from authorized capture manifests + +Status: Proposed, implementation supplied for review + +Date: 2026-10-12 + +Related: ADR-024, ADR-299, ADR-371, ADR-414, ADR-415 + +## Problem + +The published `ruvnet/wifi-densepose-pretrained` v2 encoder points to a trainer +that is absent from the current repository. The existing JavaScript trainer +produces a separate artifact family. Neither an upload date nor a model card +score establishes that a newer capture is available or suitable for training. +The published metric describes temporal embeddings, not presence or pose. + +## Decision + +Add an isolated Python CLI at `scripts/train-csi-encoder.py`. It does not change +the npm operator surface, production sensing, or a deployed model. + +1. Admit local feature captures using a versioned manifest with complete file + SHA256, declared UTC capture date, room, session, explicit partition, + real/synthetic evidence class, and affirmative permission to use the data. + Permission is an operator assertion, not independently verified consent. +2. Require disjoint rooms by default, or explicitly choose disjoint sessions. + Session separation alone is not evidence of generalization across rooms. + Reject duplicate files, copied observations across partitions, nonfinite + features, malformed records, and oversized input. Raw CSI requires a + separately reviewed feature adapter; do not fabricate features or labels. +3. Fit all standardization statistics on training observations only. Train an + 8 to 64 to 128 encoder with temporal contrastive pairs. Positives are within + two seconds of the same session and node; negatives are more than thirty + seconds away. No arbitrary batch row becomes a negative. +4. Select the checkpoint using validation only. Evaluate the final frozen + checkpoint and raw, standardized, and random encoder controls using the same + test triplets. Reusing test results for another experiment invalidates an + untouched acceptance claim and requires an external holdout policy. +5. Export safetensors, preprocessing, file hashes, actual dependency versions, + configuration, data manifest digest, and honest metrics. Refuse output + directory reuse. Failed or interrupted work cannot represent completion. +6. All runs retain `promotionAuthorized=false`. Temporal proxy improvements + never automatically qualify occupancy, vitals, pose, or an edge deployment. + +## Execution and boundaries + +ruOS can validate manifests and execute the CPU trainer. A separately +authorized Vast machine can execute the same CLI with CUDA and a protected +local dataset mount. The harness neither rents machines nor transfers data, +reads vault credentials, publishes artifacts, or promotes candidates. A +cooperative runtime limit is not a provider billing cap or instance teardown; +operators must separately bound and end any GPU rental. + +Slack can carry private handoff metadata. Public federation messages must be +limited to code coordination and public evidence. A failed coordination tool +is not a successful claim, agent execution, or training result. + +## Alternatives + +Extending the existing pose gate would conflate pose and embedding metrics. +Recovering deleted CSI from Git history conflicts with ADR-299's incident +boundary. An unrestricted GPU sweep before resolving provenance would spend +money without producing a defensible comparison. These options are rejected. + +## Validation and acceptance + +The focused unittest suite exercises actual CPU gradients, deterministic replay, +artifact loading, current collector compatibility, partition leakage, malformed +input and failure paths. Fixtures are generated temporarily and explicitly +synthetic; they establish software behavior only. No raw captures enter Git or CI. + +Real model improvement remains unverified until an authorized manifest and +adequate independent capture groups are available. A separate deployment gate +must compare downstream task performance on untouched rooms, quantization +parity, and measured device RAM and latency. A 4-bit file size alone proves +none of those properties. Keep the existing deployed model for rollback. diff --git a/docs/csi-encoder-training.md b/docs/csi-encoder-training.md new file mode 100644 index 000000000..6d81112cb --- /dev/null +++ b/docs/csi-encoder-training.md @@ -0,0 +1,104 @@ +# Reproducible CSI encoder training + +This harness trains a small temporal embedding model from explicit eight-value +CSI features. It is not a pose, presence, or vital-sign classifier. It produces +local candidates only, with no automatic publishing or promotion. + +## Set up + +Use Python 3.11 or 3.12 and an isolated environment. For the tested CPU setup: + +```sh +python3 -m venv .venv-csi +. .venv-csi/bin/activate +python -m pip install torch==2.11.0 --index-url https://download.pytorch.org/whl/cpu +python -m pip install -r scripts/requirements-csi-training.txt +PYTHONPATH=scripts python -m unittest discover -s scripts/tests -p 'test_csi_training*.py' -v +``` + +Install an appropriate CUDA build of the same PyTorch version for an authorized +GPU host. Record the resolved environment with `python -m pip freeze` alongside +the private run. CPU and GPU results need not be bit-identical. A requested CUDA +run fails when CUDA is unavailable instead of silently falling back to CPU. + +## Prepare private data + +Create a private JSON manifest with `schema_version: 1` and a `files` array. +Each entry has the following fields: + +| Field | Contract | +| --- | --- | +| `path` | Local regular file, relative to the manifest or absolute | +| `sha256` | Exact lowercase SHA256 of the complete file | +| `captured_at` | Actual declared capture time in UTC ISO 8601, not file modification or model upload time | +| `room_id` | Stable pseudonymous room identifier | +| `session_id` | Stable capture session identifier | +| `split` | `train`, `validation`, or `test` | +| `evidence` | `real` or `synthetic` | +| `use_permitted` | Explicit boolean `true`, asserted by the authorized operator | + +Keep producer revision, sensor configuration, consent, permitted execution sites, +label provenance and schema documentation with the private source manifest. +The CLI validates structural evidence; it cannot prove those operator assertions. +Never mark a capture usable solely because someone mentioned it in a message. + +Canonical JSONL rows contain `timestamp` in seconds, `node_id`, and `features` +as eight finite numbers. The explicit legacy feature stream and the current +collector feature envelope are supported. Raw subcarrier amplitudes are not +silently repurposed as features. Missing labels are never synthesized. + +Room separation is the default and requires at least three independent rooms +across training, validation and test. `--group-by session_id` allows distinct +sessions in the same room, but cannot substantiate performance in unseen rooms. +Each partition needs same-node/session positive pairs within two seconds and +negative observations more than thirty seconds apart. The training set alone +determines preprocessing statistics. + +## Inspect and train + +```sh +python scripts/train-csi-encoder.py inspect --manifest /private/csi/manifest.json +python scripts/train-csi-encoder.py train \ + --manifest /private/csi/manifest.json \ + --output /private/runs/csi-candidate-001 \ + --device cpu --epochs 20 --seed 42 --max-seconds 600 +``` + +For synthetic software experiments, add `--allow-synthetic`; results retain +the `SYNTHETIC` evidence label. Do not use those results as sensor benchmarks. +The output directory must not already exist. No input is changed or uploaded. + +For Vast, first provision an authorized isolated host, check its hourly price, +bind only the approved private dataset and pin the code commit and environment. +Then use the same command with `--device cuda`. The CLI does not access provider +keys, create instances, copy private captures, or stop the rental. Its cooperative +runtime limit must be backed by the operator's process and instance deadline. + +## Review the candidate + +The output includes `model.safetensors`, `preprocessing.json`, `metrics.json`, +and `status.json`. Check completion, manifest digest, implementation source +digests and complete file hashes. +Load preprocessing together with the model; never feed unstandardized features +directly into the network. Validation selects a checkpoint; the final test +comparison uses identical triplets for the candidate and controls. + +The loader applies the saved training statistics and verifies completion and +artifact hashes. Hashes detect corruption; they are not a trusted signature. + +```python +# Run with PYTHONPATH=scripts. raw_features has shape [N, 8]. +from csi_training.model import load_bundle + +encoder = load_bundle('/private/runs/csi-candidate-001') +embeddings = encoder.encode(raw_features) +``` + +`promotionAuthorized` remains false even if the temporal metric improves. +Temporal proximity is a proxy, and shared frame structure can make it easy. +Repeated test-driven tuning also contaminates a test set. Qualify downstream +tasks on fresh independent rooms and sessions before deployment. Quantization, +device RAM, latency, and compatibility with the production runtime remain +separate measured gates. + +See [ADR-419](adr/ADR-419-reproducible-csi-encoder-training.md) for boundaries. diff --git a/scripts/csi_training/__init__.py b/scripts/csi_training/__init__.py new file mode 100644 index 000000000..5a332b55a --- /dev/null +++ b/scripts/csi_training/__init__.py @@ -0,0 +1 @@ +"""Reproducible CSI feature training; no implicit raw CSI conversion.""" diff --git a/scripts/csi_training/data.py b/scripts/csi_training/data.py new file mode 100644 index 000000000..3a6e500d2 --- /dev/null +++ b/scripts/csi_training/data.py @@ -0,0 +1,257 @@ +"""Bounded, hash verified CSI feature intake and explicit grouped partitions. + +Manifest v1 has a ``files`` list. Each entry supplies path, sha256, +captured_at (UTC ISO 8601), room_id, session_id, split, evidence and +use_permitted. Capture dates are declarations, not inferred from file mtimes. +""" +from dataclasses import dataclass +from datetime import datetime, timezone +import hashlib +import json +import math +from pathlib import Path +import re +import stat + +import numpy as np + +SPLITS = ("train", "validation", "test") +MAX_FILE_BYTES = 256 * 1024 * 1024 +MAX_TOTAL_BYTES = 512 * 1024 * 1024 +MAX_MANIFEST_BYTES = 1024 * 1024 +MAX_LINE_BYTES = 65536 +MAX_FRAMES = 500000 +MAX_INPUT_LINES = 1000000 + + +class DataError(ValueError): + """Input does not satisfy the training evidence contract.""" + + +@dataclass +class Partition: + features: np.ndarray + timestamps: np.ndarray + node_ids: list + groups: list + session_ids: list + + +@dataclass +class Dataset: + partitions: dict + inventory: list + manifest_sha256: str + group_by: str + + +@dataclass +class Standardizer: + mean: np.ndarray + scale: np.ndarray + + def transform(self, features): + values = np.asarray(features, dtype=np.float64) + if values.ndim != 2 or values.shape[1] != 8 or not np.isfinite(values).all(): + raise DataError("standardizer requires finite Nx8 features") + result = (values - self.mean) / self.scale + if not np.isfinite(result).all(): + raise DataError("standardization overflow") + return result.astype(np.float32) + + +def fit_standardizer(train_features): + values = np.asarray(train_features, dtype=np.float64) + if values.ndim != 2 or values.shape[1] != 8 or len(values) < 2 or not np.isfinite(values).all(): + raise DataError("fit requires at least two finite training feature rows") + mean, scale = values.mean(axis=0), values.std(axis=0) + if not np.isfinite(mean).all() or not np.isfinite(scale).all(): + raise DataError("training statistics overflow") + return Standardizer(mean, np.where(scale < 1e-8, 1.0, scale)) + + +def _json(text): + def pairs(items): + result = {} + for key, value in items: + if key in result: + raise DataError("duplicate JSON key") + result[key] = value + return result + try: + return json.loads(text, object_pairs_hook=pairs, + parse_constant=lambda _: (_ for _ in ()).throw(DataError("nonfinite JSON number"))) + except (ValueError, UnicodeError) as exc: + raise DataError("invalid JSON input") from exc + + +def _regular(path, maximum): + if path.is_symlink(): + raise DataError("symlink input is not permitted") + try: + info = path.stat() + except OSError as exc: + raise DataError("input file unavailable") from exc + if not stat.S_ISREG(info.st_mode) or info.st_size > maximum: + raise DataError("input must be a bounded regular file") + return info.st_size + + +def _text(value, name): + if not isinstance(value, str) or not value.strip() or len(value) > 256 or any(ord(c) < 32 for c in value): + raise DataError(f"invalid {name}") + return value + + +def _manifest(path): + path = Path(path).absolute() + _regular(path, MAX_MANIFEST_BYTES) + raw = path.read_bytes() + if len(raw) > MAX_MANIFEST_BYTES: + raise DataError("manifest exceeds limit") + obj = _json(raw) + if not isinstance(obj, dict) or type(obj.get("schema_version")) is not int or obj["schema_version"] != 1: + raise DataError("manifest schema_version must equal 1") + entries = obj.get("files") + if not isinstance(entries, list) or not entries or len(entries) > 1024: + raise DataError("manifest files must contain 1 to 1024 entries") + checked, digests, total = [], set(), 0 + for index, entry in enumerate(entries): + if not isinstance(entry, dict): + raise DataError("manifest file entry must be an object") + for key in ("room_id", "session_id", "path"): + _text(entry.get(key), key) + digest = entry.get("sha256") + if not isinstance(digest, str) or not re.fullmatch(r"[0-9a-f]{64}", digest): + raise DataError("sha256 must be a lowercase hex digest") + if digest in digests: + raise DataError("duplicate file hash") + digests.add(digest) + if entry.get("split") not in SPLITS or entry.get("evidence") not in ("real", "synthetic"): + raise DataError("invalid split or evidence") + if entry.get("use_permitted") is not True: + raise DataError("explicit use_permitted=true is required") + captured = _text(entry.get("captured_at"), "captured_at") + try: + date = datetime.fromisoformat(captured.replace("Z", "+00:00")) + except ValueError as exc: + raise DataError("invalid captured_at") from exc + if date.tzinfo is None or date.utcoffset() != timezone.utc.utcoffset(date): + raise DataError("captured_at must specify UTC") + source = path.parent / entry["path"] + total += _regular(source, MAX_FILE_BYTES) + if total > MAX_TOTAL_BYTES: + raise DataError("total dataset exceeds byte limit") + checked.append((dict(entry, captured_at=date.isoformat()), source, index)) + return checked, hashlib.sha256(raw).hexdigest() + + +def discover(manifest): + """Inventory declared capture dates; load_manifest verifies hashes and frames.""" + entries, _ = _manifest(manifest) + return sorted([dict(captured_at=e["captured_at"], sha256=e["sha256"], + split=e["split"], evidence=e["evidence"], + room_id=e["room_id"], session_id=e["session_id"]) + for e, _, _ in entries], key=lambda e: datetime.fromisoformat(e["captured_at"]), reverse=True) + + +def _number(value): + if type(value) not in (int, float) or abs(value) > 1e20 or not math.isfinite(value): + raise DataError("numeric values must be finite and bounded") + return 0.0 if value == 0 else float(value) + + +def load_manifest(manifest, group_by="room_id", allow_synthetic=False): + """Validate partitions without deriving labels or creating raw CSI features.""" + if group_by not in ("room_id", "session_id"): + raise DataError("group_by must be room_id or session_id") + entries, manifest_digest = _manifest(manifest) + owners, session_owners, fingerprints, rows = {}, {}, {}, {s: [] for s in SPLITS} + seen_observations = set() + frame_count = 0 + input_lines = 0 + for entry, source, index in entries: + split, group = entry["split"], entry[group_by] + if entry["evidence"] == "synthetic" and not allow_synthetic: + raise DataError("synthetic data requires explicit allow_synthetic") + for mapping, key in ((owners, group), (session_owners, entry["session_id"])): + if key in mapping and mapping[key] != split: + raise DataError("group or session appears across splits") + mapping[key] = split + digest, last, count, size = hashlib.sha256(), {}, 0, 0 + with source.open("rb") as stream: + while True: + line = stream.readline(MAX_LINE_BYTES + 1) + if not line: + break + input_lines += 1 + if input_lines > MAX_INPUT_LINES: + raise DataError("dataset exceeds input line limit") + size += len(line) + if len(line) > MAX_LINE_BYTES or size > MAX_FILE_BYTES: + raise DataError("capture line or file exceeds limit") + digest.update(line) + if not line.strip(): + continue + row = _json(line) + if not isinstance(row, dict): + raise DataError("capture row must be an object") + # Mixed legacy recordings are supported only for explicit types. + if row.get("type") in ("vitals", "raw_csi"): + continue + if row.get("type") not in (None, "feature", "features"): + raise DataError("unsupported capture row type") + features = row.get("features") + node = row.get("node_id") + if isinstance(features, dict): + # CsiRecorder.write_frame removes the packet type and places + # explicit packet features/node_id in this nested envelope. + # Outer timestamp is already seconds; timestamp_us is metadata. + if row.get("type") is not None or features.get("type") not in (None, "feature", "features"): + raise DataError("unsupported collector envelope") + node = features.get("node_id") + features = features.get("features") + if not isinstance(features, list) or len(features) != 8: + raise DataError("feature rows require eight explicit features") + features = [_number(v) for v in features] + timestamp = _number(row.get("timestamp")) + if type(node) is int: + if node < 0 or node > 2**32 - 1: + raise DataError("invalid node_id") + node = str(node) + else: + node = _text(node, "node_id") + observation = (entry["session_id"], node, timestamp) + if observation in seen_observations: + raise DataError("duplicate session/node/timestamp observation") + seen_observations.add(observation) + if node in last and timestamp < last[node]: + raise DataError("timestamps must be monotonic per node") + last[node] = timestamp + # Bounded copy heuristic, not proof of absence of leakage. + # Idle/quantized vectors can legitimately recur at distinct times. + signature = json.dumps([timestamp, node, features], separators=(",", ":")).encode() + fingerprint = hashlib.sha256(signature).digest() + if fingerprint in fingerprints and fingerprints[fingerprint] != split: + raise DataError("duplicate feature frame across splits") + fingerprints[fingerprint] = split + rows[split].append((features, timestamp, node, group, entry["session_id"])) + count += 1 + frame_count += 1 + if frame_count > MAX_FRAMES: + raise DataError("dataset exceeds frame limit") + if digest.hexdigest() != entry["sha256"]: + raise DataError(f"file hash mismatch at manifest entry {index}") + if count == 0: + raise DataError("capture has no explicit feature rows; raw CSI requires a reviewed adapter") + partitions = {} + for split, values in rows.items(): + if len(values) < 2: + raise DataError(f"partition {split} requires at least two feature rows") + features, timestamps, nodes, groups, sessions = zip(*values) + partitions[split] = Partition(np.asarray(features, dtype=np.float64), + np.asarray(timestamps, dtype=np.float64), + list(nodes), list(groups), list(sessions)) + inventory = sorted([{k: e[k] for k in ("captured_at", "sha256", "split", "evidence", "room_id", "session_id")} + for e, _, _ in entries], key=lambda e: datetime.fromisoformat(e["captured_at"]), reverse=True) + return Dataset(partitions, inventory, manifest_digest, group_by) diff --git a/scripts/csi_training/model.py b/scripts/csi_training/model.py new file mode 100644 index 000000000..6a9846c33 --- /dev/null +++ b/scripts/csi_training/model.py @@ -0,0 +1,83 @@ +"""Small temporal CSI encoder; not a pose or occupancy classifier.""" +import torch +from torch import nn +from torch.nn import functional as F + + +class CSIEncoder(nn.Module): + """8 -> 64 -> 128, batch normalization, GELU and unit embeddings.""" + def __init__(self): + super().__init__() + self.w1 = nn.Linear(8, 64) + self.bn1 = nn.BatchNorm1d(64) + self.w2 = nn.Linear(64, 128) + self.bn2 = nn.BatchNorm1d(128) + + def forward(self, x): + return F.normalize(self.bn2(self.w2(F.gelu(self.bn1(self.w1(x))))), dim=-1) + + +class PreprocessedEncoder: + """Verified local candidate loader; includes train-fitted normalization. + + Hashes detect accidental artifact corruption. They are not signatures and + do not authenticate an untrusted directory or authorize model promotion. + """ + def __init__(self, directory, device='cpu'): + import hashlib + import json + from pathlib import Path + import numpy as np + from safetensors.torch import load_file + + root = Path(directory) + paths = {name: root / name for name in ('status.json', 'metrics.json', 'preprocessing.json', 'model.safetensors')} + for name, path in paths.items(): + if path.is_symlink() or not path.is_file() or path.stat().st_size > 2 * 1024 * 1024: + raise ValueError(f'Invalid bounded artifact: {name}') + status = json.loads(paths['status.json'].read_text()) + if not isinstance(status, dict) or status.get('status') != 'complete' or status.get('promotionAuthorized') is not False: + raise ValueError('Candidate lifecycle is not complete and proposal-only') + report = json.loads(paths['metrics.json'].read_text()) + if not isinstance(report, dict): + raise ValueError('Invalid metrics schema') + if report.get('architecture') != 'Linear(8,64),BN64,GELU,Linear(64,128),BN128,L2': + raise ValueError('Unsupported encoder architecture') + if report.get('status') != 'complete' or report.get('promotionAuthorized') is not False: + raise ValueError('Candidate is not complete') + for filename, key in [('model.safetensors', 'model_sha256'), ('preprocessing.json', 'preprocessing_sha256')]: + if hashlib.sha256(paths[filename].read_bytes()).hexdigest() != report.get(key): + raise ValueError(f'Artifact hash mismatch: {filename}') + prep = json.loads(paths['preprocessing.json'].read_text()) + if not isinstance(prep, dict): + raise ValueError('Invalid preprocessing schema') + mean, scale = np.asarray(prep['mean'], dtype=np.float64), np.asarray(prep['scale'], dtype=np.float64) + if (mean.shape != (8,) or scale.shape != (8,) or not np.isfinite(mean).all() + or not np.isfinite(scale).all() or not (scale > 0).all() + or prep.get('fit_partition') != 'train' or prep.get('feature_count') != 8): + raise ValueError('Invalid preprocessing contract') + self.mean, self.scale, self.device = mean, scale, device + self.model = CSIEncoder().to(device).eval() + self.model.load_state_dict(load_file(str(paths['model.safetensors']), device=device), strict=True) + if any(not torch.isfinite(v).all() for v in self.model.state_dict().values()): + raise ValueError('Non-finite model weights') + + def encode(self, features): + import numpy as np + values = np.asarray(features, dtype=np.float64) + if values.ndim != 2 or values.shape[1] != 8 or not 0 < len(values) <= 4096 or not np.isfinite(values).all(): + raise ValueError('Expected 1 to 4096 finite eight-feature rows') + with np.errstate(over='ignore', invalid='ignore'): + standardized = ((values - self.mean) / self.scale).astype(np.float32) + if not np.isfinite(standardized).all(): + raise ValueError('Input normalization overflow') + with torch.no_grad(): + result = self.model(torch.from_numpy(standardized).to(self.device)).cpu().numpy() + if not np.isfinite(result).all(): + raise ValueError('Non-finite encoder output') + return result + + +def load_bundle(directory, device='cpu'): + """Load a complete local bundle with verified hashes and preprocessing.""" + return PreprocessedEncoder(directory, device=device) diff --git a/scripts/csi_training/train.py b/scripts/csi_training/train.py new file mode 100644 index 000000000..b04d6ba59 --- /dev/null +++ b/scripts/csi_training/train.py @@ -0,0 +1,258 @@ +"""Reproducible group-held-out temporal contrastive training. + +No remote code, model upload, provisioning, or automatic promotion. Temporal +triplet accuracy describes embeddings only, never occupancy or pose accuracy. +""" +import argparse +import hashlib +import json +import math +import os +from pathlib import Path +import random +import time + +import numpy as np + + +def temporal_triplets(partition, seed, limit=4096): + """One positive 030s, within each session/node. + + No in-batch negatives: each anchor has exactly its explicitly screened + negative. This avoids silently labeling nearby windows as negative. + """ + rng = np.random.default_rng(seed) + groups = {} + for i, key in enumerate(zip(partition.session_ids, partition.node_ids)): + groups.setdefault(key, []).append(i) + candidates, eligible = [], 0 + for indices in groups.values(): + ids = np.asarray(indices) + times = np.asarray(partition.timestamps)[ids] + order = np.argsort(times, kind='stable') + ids, times = ids[order], times[order] + # Binary search bounds avoid allocating an NxN distance matrix. + for j in range(len(ids)): + t = times[j] + lo = np.searchsorted(times, t - 2, side='left') + hi = np.searchsorted(times, t + 2, side='right') + equal_lo = np.searchsorted(times, t, side='left') + equal_hi = np.searchsorted(times, t, side='right') + npos_left = equal_lo - lo + npos = npos_left + hi - equal_hi + left = np.searchsorted(times, t - 30, side='left') + right = np.searchsorted(times, t + 30, side='right') + nneg = left + len(ids) - right + if npos > 0 and nneg: + n = int(rng.integers(nneg)) + neg = ids[n] if n < left else ids[right + n - left] + draw = int(rng.integers(npos)) + p = lo + draw if draw < npos_left else equal_hi + draw - npos_left + candidate = (ids[j], ids[p], neg) + # Reservoir sampling bounds candidate storage globally while + # giving every eligible anchor the same inclusion probability. + eligible += 1 + if len(candidates) < limit: + candidates.append(candidate) + else: + slot = int(rng.integers(eligible)) + if slot < limit: + candidates[slot] = candidate + if not candidates: + raise ValueError('Each split needs same-session/node positive pairs 030s') + rng.shuffle(candidates) + return np.asarray(candidates[:limit], dtype=np.int64) + + +def triplet_accuracy(embeddings, triplets): + a, p, n = (embeddings[triplets[:, i]] for i in range(3)) + pos, neg = np.sum(a * p, axis=1), np.sum(a * n, axis=1) + # Ties earn half credit; constant embeddings cannot score 100%. + return float(np.mean((pos > neg) + 0.5 * (pos == neg))) + + +def unit_rows(x): + return x / np.maximum(np.linalg.norm(x, axis=1, keepdims=True), 1e-12) + + +def sha256(path): + digest = hashlib.sha256() + with open(path, 'rb') as f: + for block in iter(lambda: f.read(1024 * 1024), b''): + digest.update(block) + return digest.hexdigest() + + +def write_json(path, value): + path.write_text(json.dumps(value, indent=2, sort_keys=True, allow_nan=False) + '\n') + + +def run_training(args): + # Set before CUDA context creation; deterministic ops fail rather than drift. + os.environ.setdefault('CUBLAS_WORKSPACE_CONFIG', ':4096:8') + import torch + import safetensors + from safetensors.torch import save_file + from .data import load_manifest, fit_standardizer + from .model import CSIEncoder + + started = time.monotonic() + def check_budget(): + if time.monotonic() - started >= args.max_seconds: + raise TimeoutError('Training time budget exhausted; candidate is incomplete and cannot promote') + + for name, low, high in [('epochs', 1, 10000), ('batch_size', 2, 4096), + ('max_triplets', 2, 100000), ('max_seconds', 1, 86400)]: + value = getattr(args, name) + if not math.isfinite(value) or not low <= value <= high: + raise ValueError(f'{name} must be between {low} and {high}') + if not 0 < args.learning_rate <= 0.1 or not 0.01 <= args.temperature <= 10: + raise ValueError('Invalid learning rate or temperature') + if not 0 <= args.seed < 2**32: + raise ValueError('seed must be in [0, 2**32)') + out = Path(args.output) + if out.exists(): + raise ValueError('Output must not exist; never overwrite a training run') + dataset = load_manifest(args.manifest, group_by=args.group_by, allow_synthetic=args.allow_synthetic) + scaler = fit_standardizer(dataset.partitions['train'].features) + triplets = {name: temporal_triplets(part, args.seed + i, args.max_triplets) + for i, (name, part) in enumerate(dataset.partitions.items())} + values = {name: scaler.transform(part.features).astype(np.float32) + for name, part in dataset.partitions.items()} + if any(not np.isfinite(x).all() for x in values.values()): + raise ValueError('Non-finite standardized input') + if args.device == 'cuda' and not torch.cuda.is_available(): + raise ValueError('CUDA requested but unavailable; no silent CPU fallback') + random.seed(args.seed) + np.random.seed(args.seed) + torch.manual_seed(args.seed) + torch.set_num_threads(1) + torch.use_deterministic_algorithms(True) + torch.backends.cudnn.benchmark = False + model = CSIEncoder().to(args.device) + initial = {k: v.detach().cpu().clone() for k, v in model.state_dict().items()} + tensors = {k: torch.from_numpy(v).to(args.device) for k, v in values.items()} + + def encode_partition(network, tensor): + chunks = [] + for batch in tensor.split(args.batch_size): + check_budget() + chunks.append(network(batch).cpu().numpy()) + return np.concatenate(chunks, axis=0) + + def embed(network): + network.eval() + result = {} + with torch.no_grad(): + for name, tensor in tensors.items(): + check_budget() + result[name] = encode_partition(network, tensor) + return result + + # Preserve initialization; evaluate the random control and candidate test + # embeddings only after checkpoint selection is frozen. + optimizer = torch.optim.AdamW(model.parameters(), lr=args.learning_rate) + rng = np.random.default_rng(args.seed) + history, best, best_epoch, best_state = [], -1.0, 0, None + out.mkdir(parents=True, exist_ok=False) + write_json(out / 'status.json', {'status': 'running', 'promotionAuthorized': False}) + try: + for epoch in range(args.epochs): + check_budget() + model.train() + ordered = triplets['train'][rng.permutation(len(triplets['train']))] + losses = [] + for start in range(0, len(ordered), args.batch_size): + check_budget() + batch = ordered[start:start + args.batch_size] + # Concatenation gives BN at least three samples even for tail=1. + flat = torch.as_tensor(batch.reshape(-1), device=args.device) + z = model(tensors['train'][flat]).reshape(-1, 3, 128) + logits = torch.stack(((z[:, 0] * z[:, 1]).sum(-1), + (z[:, 0] * z[:, 2]).sum(-1)), dim=1) / args.temperature + loss = torch.nn.functional.cross_entropy(logits, torch.zeros(len(batch), dtype=torch.long, device=args.device)) + if not torch.isfinite(loss): + raise ValueError('Non-finite loss') + optimizer.zero_grad(set_to_none=True) + loss.backward() + torch.nn.utils.clip_grad_norm_(model.parameters(), 5, error_if_nonfinite=True) + optimizer.step() + losses.append(float(loss.detach())) + model.eval() + with torch.no_grad(): + validation = encode_partition(model, tensors['validation']) + score = triplet_accuracy(validation, triplets['validation']) + history.append({'epoch': epoch + 1, 'loss': float(np.mean(losses)), 'validation_temporal_triplet_accuracy': score}) + if score > best: + best, best_epoch = score, epoch + 1 + best_state = {k: v.detach().cpu().clone() for k, v in model.state_dict().items()} + check_budget() + model.load_state_dict(best_state) + learned = embed(model) + random_model = CSIEncoder().to(args.device) + random_model.load_state_dict(initial) + random_embeddings = embed(random_model) + if any(not np.isfinite(x).all() for x in learned.values()): + raise ValueError('Non-finite final embeddings') + metrics = {name: {'triplets': len(triplets[name]), + 'trained': triplet_accuracy(learned[name], triplets[name]), + 'random_encoder': triplet_accuracy(random_embeddings[name], triplets[name]), + 'standardized_input': triplet_accuracy(unit_rows(values[name]), triplets[name]), + 'raw_input': triplet_accuracy(unit_rows(dataset.partitions[name].features), triplets[name])} + for name in tensors} + delta = sum(float((best_state[k] - initial[k]).square().sum()) for k, _ in model.named_parameters()) + save_file(best_state, str(out / 'model.safetensors')) + write_json(out / 'preprocessing.json', {'mean': np.asarray(scaler.mean).tolist(), 'scale': np.asarray(scaler.scale).tolist(), 'fit_partition': 'train', 'feature_count': 8}) + report = {'status': 'complete', 'promotionAuthorized': False, + 'metric': 'temporal_triplet_accuracy_not_pose_or_presence', + 'architecture': 'Linear(8,64),BN64,GELU,Linear(64,128),BN128,L2', + 'parameter_count': sum(p.numel() for p in model.parameters()), + 'evidence': 'SYNTHETIC' if any(e['evidence'] == 'synthetic' for e in dataset.inventory) else 'MEASURED', + 'best_epoch': best_epoch, 'history': history, 'metrics': metrics, + 'trainable_parameter_delta_squared': delta, 'group_by': args.group_by, 'seed': args.seed, 'device': args.device, + 'config': {k: v for k, v in vars(args).items() if k not in ('manifest', 'output')}, + 'manifest_sha256': dataset.manifest_sha256, + 'source_sha256': {name: sha256(Path(__file__).parent / name) + for name in ('train.py', 'data.py', 'model.py')}, + 'data_files': [{'sha256': e['sha256'], 'split': e['split'], 'evidence': e['evidence'], 'captured_at': e['captured_at']} for e in dataset.inventory], + 'model_sha256': sha256(out / 'model.safetensors'), + 'preprocessing_sha256': sha256(out / 'preprocessing.json'), + 'versions': {'torch': torch.__version__, 'numpy': np.__version__, 'safetensors': safetensors.__version__}, + 'elapsed_seconds': time.monotonic() - started, + 'limitations': ['Temporal proxy does not establish task accuracy', 'No INT4 export or hardware RAM/latency claim', 'No automatic promotion or upload', 'Time budget is cooperative between batches, use an external process timeout for a hard deadline', 'Repeated test reuse requires an external frozen holdout policy']} + write_json(out / 'metrics.json', report) + write_json(out / 'status.json', {'status': 'complete', 'promotionAuthorized': False}) + return report + except BaseException: + write_json(out / 'status.json', {'status': 'failed', 'promotionAuthorized': False}) + raise + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + sub = parser.add_subparsers(dest='command', required=True) + inspect = sub.add_parser('inspect', help='Validate local manifest without training') + train = sub.add_parser('train', help='Train a local candidate; never publish or promote') + for command in (inspect, train): + command.add_argument('--manifest', required=True) + command.add_argument('--group-by', choices=['room_id', 'session_id'], default='room_id') + command.add_argument('--allow-synthetic', action='store_true') + train.add_argument('--output', required=True) + train.add_argument('--device', choices=['cpu', 'cuda'], default='cpu') + train.add_argument('--epochs', type=int, default=20) + train.add_argument('--batch-size', type=int, default=64) + train.add_argument('--max-triplets', type=int, default=4096) + train.add_argument('--seed', type=int, default=42) + train.add_argument('--max-seconds', type=float, default=600) + train.add_argument('--learning-rate', type=float, default=0.001) + train.add_argument('--temperature', type=float, default=0.1) + args = parser.parse_args(argv) + try: + if args.command == 'inspect': + from .data import load_manifest + result = load_manifest(args.manifest, group_by=args.group_by, allow_synthetic=args.allow_synthetic).inventory + else: + result = run_training(args) + print(json.dumps(result, indent=2, allow_nan=False)) + except (ValueError, OSError, TimeoutError, ImportError) as exc: + parser.exit(2, f'error: {exc}\n') diff --git a/scripts/requirements-csi-training.txt b/scripts/requirements-csi-training.txt new file mode 100644 index 000000000..230355f31 --- /dev/null +++ b/scripts/requirements-csi-training.txt @@ -0,0 +1,5 @@ +# Replayed CPU integration environment. Install the appropriate CPU/CUDA torch +# wheel using an explicit index URL; actual versions are recorded in each run. +numpy==2.4.4 +torch==2.11.0 +safetensors==0.8.0 diff --git a/scripts/tests/test_csi_training_data.py b/scripts/tests/test_csi_training_data.py new file mode 100644 index 000000000..2d1d5cd49 --- /dev/null +++ b/scripts/tests/test_csi_training_data.py @@ -0,0 +1,222 @@ +"""Synthetic fixture contract tests, no physical sensing claims.""" +import hashlib +import importlib.util +import struct +import json +from pathlib import Path +import sys +import tempfile +import unittest +from unittest.mock import patch + +import numpy as np + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from csi_training.data import DataError, discover, fit_standardizer, load_manifest + + +class DataContractTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.base = Path(self.temp.name) + self.manifest = self.base / "manifest.json" + self.entries = [] + for i, split in enumerate(("train", "validation", "test")): + self.write_capture(i, [dict(type="feature", timestamp=t, node_id=1, + features=[i * 10 + t + j / 10 for j in range(8)]) + for t in (0, 1, 40)]) + self.entries.append(dict(path=f"{i}.jsonl", sha256=self.sha(i), + captured_at=f"2026-10-{10+i}T00:00:00Z", + room_id=f"room{i}", session_id=f"session{i}", + split=split, evidence="synthetic", use_permitted=True)) + self.save() + + def write_capture(self, i, rows): + (self.base / f"{i}.jsonl").write_text("".join(json.dumps(r) + "\n" for r in rows)) + + def sha(self, i): + return hashlib.sha256((self.base / f"{i}.jsonl").read_bytes()).hexdigest() + + def save(self): + self.manifest.write_text(json.dumps(dict(schema_version=1, files=self.entries))) + + def load(self): + return load_manifest(self.manifest, allow_synthetic=True) + + def test_load_and_train_only_standardization(self): + data = self.load() + stats = fit_standardizer(data.partitions["train"].features) + np.testing.assert_allclose(stats.transform(data.partitions["train"].features).mean(0), 0, atol=1e-6) + self.assertGreater(stats.transform(data.partitions["test"].features).mean(), 0.5) + self.assertEqual(data.partitions["train"].session_ids, ["session0"] * 3) + self.assertEqual(discover(self.manifest)[0]["session_id"], "session2") + self.assertNotIn("path", data.inventory[0]) + + def test_synthetic_requires_opt_in(self): + with self.assertRaisesRegex(DataError, "synthetic"): + load_manifest(self.manifest) + + def test_hash_tampering(self): + with (self.base / "0.jsonl").open("a") as f: + f.write("\n") + with self.assertRaisesRegex(DataError, "hash mismatch"): + self.load() + + def test_group_leakage(self): + self.entries[1]["room_id"] = "room0" + self.save() + with self.assertRaisesRegex(DataError, "across splits"): + self.load() + load_manifest(self.manifest, group_by="session_id", allow_synthetic=True) + + def test_session_leakage_even_when_room_names_differ(self): + self.entries[1]["session_id"] = "session0" + self.save() + with self.assertRaisesRegex(DataError, "across splits"): + self.load() + + def test_duplicate_hash(self): + self.entries[1]["sha256"] = self.entries[0]["sha256"] + self.save() + with self.assertRaisesRegex(DataError, "duplicate file"): + self.load() + + def test_copied_frames_with_different_session_ids(self): + rows = [dict(timestamp=t, node_id=1, features=[t+j/10 for j in range(8)]) for t in (0, 1, 40)] + self.write_capture(1, rows) + self.entries[1]["sha256"] = self.sha(1) + self.save() + with self.assertRaisesRegex(DataError, "duplicate feature"): + self.load() + + def test_collector_plural_features_shape(self): + rows = [dict(type="features", node_id=1, seq=i, timestamp_us=t * 1000000, + timestamp=t, features=[t+j/10 for j in range(8)], + subcarriers=[t+j/10 for j in range(8)], rssi=0.0, noise_floor=0.0) + for i, t in enumerate((0, 1, 40))] + self.write_capture(0, rows) + self.entries[0]["sha256"] = self.sha(0) + self.save() + self.assertEqual(len(self.load().partitions["train"].features), 3) + + def test_actual_collector_recording_envelope(self): + spec = importlib.util.spec_from_file_location("csi_collector", Path(__file__).resolve().parents[1] / "collect-training-data.py") + collector = importlib.util.module_from_spec(spec) + spec.loader.exec_module(collector) + recorder = collector.CsiRecorder(str(self.base), "synthetic-test") + recorder.open() + try: + for i, timestamp in enumerate((0, 1, 40)): + packet = struct.pack(collector.FEATURE_PKT_FMT, collector.MAGIC_FEATURES, + 1, 0, i, timestamp * 1000000, + *[timestamp+j/10 for j in range(8)]) + recorder.write_frame(collector._parse_feature_packet(packet)) + finally: + recorder.close() + self.entries[0]["path"] = recorder.file_path.name + self.entries[0]["sha256"] = hashlib.sha256(recorder.file_path.read_bytes()).hexdigest() + self.save() + partition = self.load().partitions["train"] + self.assertEqual(partition.timestamps.tolist(), [0, 1, 40]) + self.assertEqual(partition.node_ids, ["1"] * 3) + np.testing.assert_allclose(partition.features[1], [1+j/10 for j in range(8)], atol=1e-6) + + def test_raw_collector_envelope_rejected(self): + self.write_capture(0, [dict(timestamp=1, subcarriers=[1]*8, rssi=0, + noise_floor=0, features=dict(node_id=1, seq=0))]) + self.entries[0]["sha256"] = self.sha(0) + self.save() + with self.assertRaisesRegex(DataError, "eight explicit features"): + self.load() + + def test_constant_features_at_different_times_are_not_copied_frames(self): + for i in range(3): + self.write_capture(i, [dict(timestamp=i*100+t, node_id=1, features=[0]*8) + for t in (0, 1, 40)]) + self.entries[i]["sha256"] = self.sha(i) + self.save() + self.assertEqual(len(self.load().partitions["test"].features), 3) + + def test_skipped_input_lines_are_bounded(self): + with (self.base / "0.jsonl").open("a") as f: + f.write('\n' + json.dumps(dict(type="vitals")) + '\n') + self.entries[0]["sha256"] = self.sha(0) + self.save() + with patch("csi_training.data.MAX_INPUT_LINES", 4): + with self.assertRaisesRegex(DataError, "input line limit"): + self.load() + + def test_invalid_rows(self): + cases = [dict(timestamp=1, node_id=1, features=[True] * 8), + dict(timestamp=1, node_id=1, features=[float("nan")] * 8), + dict(timestamp=1, node_id=1, features=[1] * 7), + dict(type="raw_csi", timestamp=1, node_id=1, iq_hex="0011")] + for row in cases: + with self.subTest(row=row): + self.write_capture(0, [row]) + self.entries[0]["sha256"] = self.sha(0) + self.save() + with self.assertRaises(DataError): + self.load() + + def test_duplicate_observation_same_file(self): + for conflicting in (False, True): + with self.subTest(conflicting=conflicting): + self.write_capture(0, [dict(timestamp=0, node_id=1, features=[0]*8), + dict(timestamp=0, node_id=1, features=[int(conflicting)]*8)]) + self.entries[0]["sha256"] = self.sha(0) + self.save() + with self.assertRaisesRegex(DataError, "duplicate session/node/timestamp"): + self.load() + + def test_duplicate_observation_across_files_same_session(self): + self.write_capture(3, [dict(timestamp=0, node_id=1, features=[999]*8)]) + self.entries.append(dict(self.entries[0], path="3.jsonl", sha256=self.sha(3))) + self.save() + with self.assertRaisesRegex(DataError, "duplicate session/node/timestamp"): + self.load() + + def test_independent_nodes_at_same_timestamp_allowed(self): + self.write_capture(0, [dict(timestamp=0, node_id=node, features=[node]*8) + for node in (1, 2, 3)]) + self.entries[0]["sha256"] = self.sha(0) + self.save() + self.assertEqual(len(self.load().partitions["train"].features), 3) + + def test_timestamp_order(self): + self.write_capture(0, [dict(timestamp=t, node_id=1, features=[t]*8) for t in (2, 1)]) + self.entries[0]["sha256"] = self.sha(0) + self.save() + with self.assertRaisesRegex(DataError, "monotonic"): + self.load() + + def test_permissions_and_dates(self): + for field, value in (("use_permitted", False), ("captured_at", "2026-01-01")): + original = self.entries[0][field] + self.entries[0][field] = value + self.save() + with self.assertRaises(DataError): + self.load() + self.entries[0][field] = original + + def test_symlink_and_nonregular(self): + (self.base / "link.jsonl").symlink_to(self.base / "0.jsonl") + self.entries[0]["path"] = "link.jsonl" + self.save() + with self.assertRaisesRegex(DataError, "symlink"): + self.load() + self.entries[0]["path"] = "." + self.save() + with self.assertRaisesRegex(DataError, "regular"): + self.load() + + def test_missing_partition(self): + self.entries.pop() + self.save() + with self.assertRaisesRegex(DataError, "partition test"): + self.load() + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/tests/test_csi_training_train.py b/scripts/tests/test_csi_training_train.py new file mode 100644 index 000000000..c07ab6751 --- /dev/null +++ b/scripts/tests/test_csi_training_train.py @@ -0,0 +1,156 @@ +"""Synthetic CPU integration tests, never evidence of sensing performance.""" +import argparse +import hashlib +import importlib.util +import json +from pathlib import Path +import tempfile +import unittest + +import numpy as np +from csi_training.data import Partition +from csi_training.train import temporal_triplets, triplet_accuracy, run_training + + +class SamplingTests(unittest.TestCase): + def test_triplets_respect_session_node_and_time(self): + t = np.tile(np.array([0, 1, 40, 41, 80, 81.]), 4) + nodes = ['a'] * 6 + ['b'] * 6 + ['a'] * 6 + ['b'] * 6 + sessions = ['one'] * 12 + ['two'] * 12 + partition = Partition(np.ones((24, 8)), t, nodes, ['room'] * 24, sessions) + triplets = temporal_triplets(partition, 12) + self.assertTrue(np.array_equal(triplets, temporal_triplets(partition, 12))) + for a, p, n in triplets: + self.assertNotEqual(a, p) + self.assertEqual(sessions[a], sessions[p]) + self.assertEqual(sessions[a], sessions[n]) + self.assertEqual(nodes[a], nodes[p]) + self.assertEqual(nodes[a], nodes[n]) + self.assertGreater(abs(t[a] - t[p]), 0) + self.assertLessEqual(abs(t[a] - t[p]), 2) + self.assertGreater(abs(t[a] - t[n]), 30) + + def test_duplicate_timestamp_only_pairs_rejected(self): + times = np.array([0, 0, 40, 40, 80, 80.]) + part = Partition(np.tile(np.eye(3, 8), (2, 1)), times, + ['a'] * 6, ['r'] * 6, ['s'] * 6) + with self.assertRaisesRegex(ValueError, '0