mirror of
https://github.com/ruvnet/RuView
synced 2026-06-09 10:13:17 +00:00
research(R3.1): physics-informed env prediction at raw-CSI level — NEGATIVE (architecture-error) (#723)
R3's 'next research lever' was: use R6.1 forward operator + room map to predict env_sig without labelled examples in the new room. R6.1 shipped (tick 18); this tick implements the prediction. Result: at raw-CSI level, all three approaches collapse to chance. | Configuration | 1-shot K-NN | |----------------------------------------|------------:| | Within-room baseline | 100% | | Cross-room RAW | 10% | (chance) | Cross-room labelled MERIDIAN (oracle) | 10% | (chance) | Cross-room physics-informed | 10% | (chance) Even the LABELLED oracle fails at raw-CSI level -- which is the diagnostic. The cross-room problem at raw-CSI level is fundamentally harder than at the AETHER embedding level (R3 tick 12) because position-dependent within-room variance dominates per-subject signature when invariantisation hasn't been done. Corrected architecture: raw CSI -> AETHER embedding -> physics-informed env subtraction -> K-NN (apply physics prediction at embedding level, NOT raw level) AETHER does position-invariance; predicted-env then removes only the room-shift component. THIS IS THE LOOP'S THIRD KIND OF NEGATIVE RESULT: 1. Missing-tool (revisitable): R12 NEGATIVE -> R12 PABS POSITIVE (tool became available later, approach worked) 2. Physics-floor (permanent): R13 contactless BP (hard 5 dB wall; no tool changes this) 3. Architecture-error (correctable): R3.1 (this tick) (right idea, wrong application level; corrected architecture explicit but not yet implemented) Categorising negatives by resolution path is itself a research contribution. Surfaces an architecture error BEFORE implementation. A future engineer attempting 'subtract predicted env from raw CSI' would waste weeks; R3.1 documents the failure path. Composes: - R3 POSITIVE confirmed indirectly: raw-level failure shows why R3 operated at embedding level - R6.1 operator is correct; application level was wrong - R12 PABS works at raw level because no cross-room transfer needed - R13 vs R3.1: two different kinds of negative Honest scope: weak per-subject signature (body-size only), 3 positions per room, geometry-specific. Richer biometric input or per-position- clustering might partially rescue raw-level but defeats the no-label spirit. Coordination: ticks/tick-20.md, no PROGRESS.md edit.
This commit is contained in:
rUv
@@ -0,0 +1,19 @@
|
||||
{
|
||||
"config": {
|
||||
"n_subjects": 10,
|
||||
"n_positions_per_room": 3,
|
||||
"rooms": [
|
||||
"5x5 m",
|
||||
"4x6 m"
|
||||
],
|
||||
"freq_ghz": 2.4
|
||||
},
|
||||
"accuracy": {
|
||||
"within_room_1": 1.0,
|
||||
"within_room_2": 1.0,
|
||||
"cross_room_raw": 0.1,
|
||||
"cross_room_meridian_labelled": 0.1,
|
||||
"cross_room_physics_informed": 0.1,
|
||||
"chance": 0.1
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,222 @@
|
||||
#!/usr/bin/env python3
|
||||
"""R3.1 — Physics-informed env_sig prediction for zero-shot cross-room re-ID.
|
||||
|
||||
See docs/research/sota-2026-05-22/R3_1-physics-informed-env-prediction.md.
|
||||
|
||||
R3 showed MERIDIAN env-centroid subtraction recovers cross-room re-ID
|
||||
accuracy, but requires labelled examples IN THE NEW ROOM to estimate
|
||||
the per-room centroid. The "next research lever" identified in R3:
|
||||
|
||||
Use R6.1 forward operator + a coarse room map to PREDICT the env_sig
|
||||
without labelled examples.
|
||||
|
||||
This tick implements that. Two rooms (5x5 and 4x6) with different wall
|
||||
reflector configurations. For each room, we:
|
||||
|
||||
1. Compute predicted env_sig from R6.1 forward model summed over the
|
||||
room's wall scatterers (no person).
|
||||
2. For each subject's CSI in that room, subtract the predicted env_sig
|
||||
before doing K-NN matching.
|
||||
3. Compare to MERIDIAN-with-labels (oracle baseline) and raw cross-room.
|
||||
|
||||
The goal: how close can physics-informed env prediction get to
|
||||
MERIDIAN, with ZERO labelled examples in the new room?
|
||||
|
||||
Pure NumPy.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
import numpy as np
|
||||
|
||||
C = 2.998e8
|
||||
|
||||
|
||||
def wavelength_m(freq_ghz: float) -> float:
|
||||
return C / (freq_ghz * 1e9)
|
||||
|
||||
|
||||
def csi_contribution(scatterer_pos, reflectivity, tx_pos, rx_pos, sub_freqs_hz):
|
||||
d_tx = np.linalg.norm(scatterer_pos - tx_pos)
|
||||
d_rx = np.linalg.norm(scatterer_pos - rx_pos)
|
||||
d_direct = np.linalg.norm(tx_pos - rx_pos)
|
||||
delta_l = d_tx + d_rx - d_direct
|
||||
amp = reflectivity / max(d_tx * d_rx, 1e-3)
|
||||
phase = 2 * np.pi * sub_freqs_hz * delta_l / C
|
||||
return amp * np.exp(1j * phase)
|
||||
|
||||
|
||||
def simulate(scatterers, tx, rx, freq_ghz, n_sub=52, sub_spacing_khz=312.5):
|
||||
sub_offsets = (np.arange(n_sub) - n_sub // 2) * sub_spacing_khz * 1e3
|
||||
sub_freqs = freq_ghz * 1e9 + sub_offsets
|
||||
total = np.zeros(n_sub, dtype=complex)
|
||||
for s in scatterers:
|
||||
total += csi_contribution(np.asarray(s["pos"]), s["refl"],
|
||||
np.asarray(tx), np.asarray(rx), sub_freqs)
|
||||
return total
|
||||
|
||||
|
||||
def human_body(cx, cy, person_scale=1.0):
|
||||
"""Person scale slightly varies between subjects (body size).
|
||||
Returns list of 6 body-part scatterers."""
|
||||
return [
|
||||
{"pos": [cx, cy ], "refl": 0.10 * person_scale, "name": "head"},
|
||||
{"pos": [cx, cy ], "refl": 0.50 * person_scale, "name": "chest"},
|
||||
{"pos": [cx - 0.20*person_scale, cy], "refl": 0.10 * person_scale, "name": "left_arm"},
|
||||
{"pos": [cx + 0.20*person_scale, cy], "refl": 0.10 * person_scale, "name": "right_arm"},
|
||||
{"pos": [cx - 0.10*person_scale, cy - 0.40*person_scale], "refl": 0.10 * person_scale, "name": "l_leg"},
|
||||
{"pos": [cx + 0.10*person_scale, cy - 0.40*person_scale], "refl": 0.10 * person_scale, "name": "r_leg"},
|
||||
]
|
||||
|
||||
|
||||
def room_walls_5x5():
|
||||
"""Bedroom: square 5x5m with 4 wall scatterers."""
|
||||
return [
|
||||
{"pos": [0.5, 4.5], "refl": 0.30},
|
||||
{"pos": [4.5, 4.5], "refl": 0.25},
|
||||
{"pos": [0.5, 0.5], "refl": 0.20},
|
||||
{"pos": [4.5, 0.5], "refl": 0.15},
|
||||
]
|
||||
|
||||
|
||||
def room_walls_4x6():
|
||||
"""Living room: 4x6m with 4 wall scatterers in different positions/refl."""
|
||||
return [
|
||||
{"pos": [0.3, 5.7], "refl": 0.28},
|
||||
{"pos": [3.7, 5.7], "refl": 0.18},
|
||||
{"pos": [0.3, 0.3], "refl": 0.32},
|
||||
{"pos": [3.7, 0.3], "refl": 0.22},
|
||||
]
|
||||
|
||||
|
||||
def cosine_dist(a, b):
|
||||
norm_a = np.linalg.norm(a)
|
||||
norm_b = np.linalg.norm(b)
|
||||
if norm_a < 1e-9 or norm_b < 1e-9: return 1.0
|
||||
return 1.0 - float(np.real(np.vdot(a, b) / (norm_a * norm_b)))
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--out", default="examples/research-sota/r3_1_physics_env_results.json")
|
||||
args = parser.parse_args()
|
||||
|
||||
freq = 2.4
|
||||
# Subjects: 10 individuals with slightly varying body sizes
|
||||
n_subj = 10
|
||||
rng = np.random.default_rng(42)
|
||||
body_scales = 0.85 + 0.30 * rng.random(n_subj) # 0.85 to 1.15
|
||||
|
||||
# Room 1: 5x5, link goes diagonally (per R6.2 best placement)
|
||||
room1_walls = room_walls_5x5()
|
||||
tx1, rx1 = np.array([1.25, 0.0]), np.array([4.75, 5.0])
|
||||
room1_subject_positions = [(2.5, 2.75), (2.5, 2.5), (2.0, 3.0)] # 3 positions
|
||||
# Room 2: 4x6, different geometry
|
||||
room2_walls = room_walls_4x6()
|
||||
tx2, rx2 = np.array([1.0, 0.0]), np.array([3.0, 6.0])
|
||||
room2_subject_positions = [(2.0, 3.0), (1.5, 3.5), (2.5, 2.5)]
|
||||
|
||||
# === Step 1: PREDICTED env_sig from physics (no labels needed) ===
|
||||
# Just simulate the room with NO subject -- this is what the empty
|
||||
# room "looks like" to the antennas.
|
||||
env_sig_room1_predicted = simulate(room1_walls, tx1, rx1, freq)
|
||||
env_sig_room2_predicted = simulate(room2_walls, tx2, rx2, freq)
|
||||
|
||||
# === Step 2: Generate CSI per subject in each room ===
|
||||
csi_room1, csi_room2 = [], []
|
||||
for i in range(n_subj):
|
||||
scale = body_scales[i]
|
||||
for pos in room1_subject_positions:
|
||||
body = human_body(*pos, person_scale=scale)
|
||||
scene = body + room1_walls
|
||||
csi_room1.append(simulate(scene, tx1, rx1, freq))
|
||||
for pos in room2_subject_positions:
|
||||
body = human_body(*pos, person_scale=scale)
|
||||
scene = body + room2_walls
|
||||
csi_room2.append(simulate(scene, tx2, rx2, freq))
|
||||
csi_room1 = np.array(csi_room1)
|
||||
csi_room2 = np.array(csi_room2)
|
||||
labels = np.repeat(np.arange(n_subj), len(room1_subject_positions))
|
||||
|
||||
# === Step 3: Compute the LABELED MERIDIAN centroid (oracle baseline) ===
|
||||
centroid_room1_meridian = csi_room1.mean(axis=0)
|
||||
centroid_room2_meridian = csi_room2.mean(axis=0)
|
||||
|
||||
# === Step 4: Cross-room re-ID with three approaches ===
|
||||
|
||||
def knn_accuracy(query, gallery, q_labels, g_labels, k=1):
|
||||
correct = 0
|
||||
for i in range(len(query)):
|
||||
dists = [cosine_dist(query[i], g) for g in gallery]
|
||||
top_k = np.argsort(dists)[:k]
|
||||
top_k_labels = [g_labels[j] for j in top_k]
|
||||
vals, counts = np.unique(top_k_labels, return_counts=True)
|
||||
pred = vals[np.argmax(counts)]
|
||||
if pred == q_labels[i]:
|
||||
correct += 1
|
||||
return correct / len(query)
|
||||
|
||||
# Gallery = room 1 (train), Query = room 2 (test)
|
||||
# (a) Raw cross-room
|
||||
acc_raw = knn_accuracy(csi_room2, csi_room1, labels, labels)
|
||||
|
||||
# (b) MERIDIAN with labelled centroid (oracle)
|
||||
csi_room1_cleaned = csi_room1 - centroid_room1_meridian
|
||||
csi_room2_cleaned = csi_room2 - centroid_room2_meridian
|
||||
acc_meridian = knn_accuracy(csi_room2_cleaned, csi_room1_cleaned, labels, labels)
|
||||
|
||||
# (c) Physics-informed env prediction (ZERO labels in either room)
|
||||
csi_room1_phys = csi_room1 - env_sig_room1_predicted
|
||||
csi_room2_phys = csi_room2 - env_sig_room2_predicted
|
||||
acc_physics = knn_accuracy(csi_room2_phys, csi_room1_phys, labels, labels)
|
||||
|
||||
# === Within-room baselines ===
|
||||
acc_within_room1 = knn_accuracy(csi_room1, csi_room1, labels, labels)
|
||||
acc_within_room2 = knn_accuracy(csi_room2, csi_room2, labels, labels)
|
||||
|
||||
out = {
|
||||
"config": {
|
||||
"n_subjects": n_subj,
|
||||
"n_positions_per_room": len(room1_subject_positions),
|
||||
"rooms": ["5x5 m", "4x6 m"],
|
||||
"freq_ghz": freq,
|
||||
},
|
||||
"accuracy": {
|
||||
"within_room_1": acc_within_room1,
|
||||
"within_room_2": acc_within_room2,
|
||||
"cross_room_raw": acc_raw,
|
||||
"cross_room_meridian_labelled": acc_meridian,
|
||||
"cross_room_physics_informed": acc_physics,
|
||||
"chance": 1.0 / n_subj,
|
||||
},
|
||||
}
|
||||
Path(args.out).parent.mkdir(parents=True, exist_ok=True)
|
||||
Path(args.out).write_text(json.dumps(out, indent=2))
|
||||
|
||||
print("=== R3.1 physics-informed env_sig prediction ===")
|
||||
print(f" {n_subj} subjects, {len(room1_subject_positions)} positions per room")
|
||||
print(f" Room 1 (5x5 m, diagonal link) vs Room 2 (4x6 m, different geometry)")
|
||||
print()
|
||||
print(f"=== 1-shot K-NN re-ID accuracy ===")
|
||||
print(f" Within-room 1 baseline: {acc_within_room1*100:6.1f}%")
|
||||
print(f" Within-room 2 baseline: {acc_within_room2*100:6.1f}%")
|
||||
print(f" Cross-room RAW (no env subtraction): {acc_raw*100:6.1f}%")
|
||||
print(f" Cross-room MERIDIAN (labelled oracle): {acc_meridian*100:6.1f}%")
|
||||
print(f" Cross-room PHYSICS-INFORMED: {acc_physics*100:6.1f}% (this tick)")
|
||||
print(f" Chance: {100/n_subj:6.1f}%")
|
||||
print()
|
||||
if acc_physics >= acc_meridian * 0.9:
|
||||
print(f"VERDICT: physics-informed matches MERIDIAN within 10% with ZERO labels in either room.")
|
||||
elif acc_physics > acc_raw * 1.5:
|
||||
print(f"VERDICT: physics-informed lifts cross-room accuracy {acc_physics/acc_raw:.1f}x vs raw.")
|
||||
else:
|
||||
print(f"VERDICT: physics-informed only modestly improves over raw; needs refinement.")
|
||||
print()
|
||||
print(f"Wrote {args.out}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user