EDMA reproducibility package: simulation code, result data, and manuscript figures
This commit is contained in:
KiHoLee
@@ -0,0 +1,333 @@
|
||||
"""
|
||||
Merged real-data comparison figure (replaces separate Figs 4 and 5).
|
||||
===================================================================
|
||||
Evaluates ALL schemes on the cached real BERT (text) + ViT (image)
|
||||
embedding pairs (16 pairs, d = 768, measured mean affinity ~0.028)
|
||||
under the manuscript's complex block-Rayleigh channel:
|
||||
|
||||
r = h1 M1 e1 + h2 M2 e2 + n, n ~ CN(0, sigma^2 I), h_u ~ CN(0,1),
|
||||
per-block energy E_b = 1, rho = 1/sigma^2 (per-block SNR).
|
||||
|
||||
Schemes:
|
||||
1. EDMA : per-realisation Haar-mixture masks with the per-pair
|
||||
measured beta_i, closed-form demux (13).
|
||||
2. OMA : equivalent-bandwidth model, noise std x sqrt(2).
|
||||
3. Genie SIC : perfect removal of the other user's waveform.
|
||||
4. Attention : retrained reproduction of the learned predecessor,
|
||||
d = 768, trained on parametric pairs at the measured
|
||||
mean affinity with Rayleigh channels and
|
||||
channel-equalised matched-filter inputs
|
||||
x_u = Re(M_u^T r / h_u); evaluated on the REAL pairs.
|
||||
5. ToDMA-adapted: OMP sparse coding of the real embedding (T = 16
|
||||
atoms, V = 1024), T slots x L = 48 signatures,
|
||||
per-slot OMP detection on the complex observation,
|
||||
genie association, true coefficients granted.
|
||||
|
||||
Outputs: fig/fig_bertvit_merged.pdf, data/bertvit_merged.csv.
|
||||
200 fading realisations per pair -> 3,200 Monte-Carlo samples per SNR.
|
||||
Seed fixed.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import csv
|
||||
import math
|
||||
import pickle
|
||||
import time
|
||||
from pathlib import Path
|
||||
import numpy as np
|
||||
import torch
|
||||
import matplotlib
|
||||
matplotlib.use("Agg")
|
||||
import matplotlib.pyplot as plt
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
DATA = ROOT / "data"
|
||||
FIG = ROOT / "fig"
|
||||
|
||||
plt.rcParams.update({
|
||||
"font.family": "serif",
|
||||
"font.serif": ["DejaVu Serif", "Times New Roman"],
|
||||
"font.size": 9, "axes.labelsize": 9, "legend.fontsize": 6.6,
|
||||
"xtick.labelsize": 8, "ytick.labelsize": 8,
|
||||
"axes.grid": True, "grid.linestyle": "--", "grid.linewidth": 0.4,
|
||||
"grid.alpha": 0.6, "lines.linewidth": 1.4, "lines.markersize": 4.0,
|
||||
"figure.figsize": (3.15, 2.36), "pdf.fonttype": 42,
|
||||
})
|
||||
AXES_RECT = dict(left=0.205, right=0.965, top=0.955, bottom=0.185)
|
||||
|
||||
SEED = 2026
|
||||
rng = np.random.default_rng(SEED)
|
||||
torch.manual_seed(SEED)
|
||||
|
||||
D = 768
|
||||
SNRS = np.arange(0.0, 31.0, 5.0)
|
||||
NFADE = 200 # fading realisations per pair
|
||||
|
||||
|
||||
def haar(d):
|
||||
G = rng.standard_normal((d, d))
|
||||
Q, R = np.linalg.qr(G)
|
||||
return Q * np.sign(np.diag(R))
|
||||
|
||||
|
||||
def unit(v):
|
||||
return v / np.linalg.norm(v)
|
||||
|
||||
|
||||
def cosine(a, b):
|
||||
return float(abs(np.vdot(a, b)) / (np.linalg.norm(a) * np.linalg.norm(b)))
|
||||
|
||||
|
||||
def load_pairs():
|
||||
a, b = pickle.load(open(DATA / "bert_vit_cached.pkl", "rb"))
|
||||
a = np.stack([unit(x - x.mean()) for x in a])
|
||||
b = np.stack([unit(x - x.mean()) for x in b])
|
||||
betas = np.abs((a * b).sum(1))
|
||||
print(f"[pairs] {len(a)} cached BERT/ViT pairs, d={a.shape[1]}, "
|
||||
f"beta mean {betas.mean():.4f} std {betas.std():.4f}")
|
||||
return a, b, betas
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# attention model: trained at the measured mean affinity, d=768,
|
||||
# Rayleigh channels, channel-equalised MF inputs
|
||||
# ------------------------------------------------------------------
|
||||
EPS_EQ = 0.1 # regularised equalisation h*/(|h|^2+EPS_EQ):
|
||||
# caps deep-fade amplification for the learned readout
|
||||
|
||||
|
||||
def train_attention(beta0, epochs=150, steps=20, batch=48, lr=5e-4,
|
||||
l1=1.0, l2=0.5, l3=0.5):
|
||||
print(f"=== training attention reproduction (d={D}, beta={beta0:.3f}, "
|
||||
f"{epochs} epochs, Rayleigh) ===", flush=True)
|
||||
gen = torch.Generator().manual_seed(SEED)
|
||||
g0 = math.sqrt(1.0 - beta0**2)
|
||||
|
||||
def torch_pairs(n):
|
||||
e1 = torch.nn.functional.normalize(
|
||||
torch.randn(n, D, generator=gen), dim=1)
|
||||
w = torch.randn(n, D, generator=gen)
|
||||
w = w - (w * e1).sum(1, keepdim=True) * e1
|
||||
w = torch.nn.functional.normalize(w, dim=1)
|
||||
return e1, beta0 * e1 + g0 * w
|
||||
|
||||
M1 = torch.nn.Parameter(torch.linalg.qr(
|
||||
torch.randn(D, D, generator=gen))[0])
|
||||
M2 = torch.nn.Parameter(beta0 * M1.detach()
|
||||
+ g0 * torch.linalg.qr(
|
||||
torch.randn(D, D, generator=gen))[0])
|
||||
Q1 = torch.nn.Parameter(torch.randn(D, D, generator=gen) / math.sqrt(D))
|
||||
Q2 = torch.nn.Parameter(torch.randn(D, D, generator=gen) / math.sqrt(D))
|
||||
opt = torch.optim.Adam([M1, M2, Q1, Q2], lr=lr)
|
||||
eye = torch.eye(D)
|
||||
|
||||
t0 = time.time()
|
||||
for ep in range(epochs):
|
||||
for _ in range(steps):
|
||||
e1, e2 = torch_pairs(batch)
|
||||
snr_db = 5.0 + 20.0 * torch.rand(batch, 1, generator=gen)
|
||||
sig = 10 ** (-snr_db / 20.0)
|
||||
hr = torch.randn(batch, 2, generator=gen)
|
||||
hi = torch.randn(batch, 2, generator=gen)
|
||||
# complex channel on real signals; equalised MF real part:
|
||||
# x_u = Re(M_u^T r / h_u); build via real/imag components
|
||||
s1 = e1 @ M1.T
|
||||
s2 = e2 @ M2.T
|
||||
nr = sig * torch.randn(batch, D, generator=gen) / math.sqrt(2)
|
||||
ni = sig * torch.randn(batch, D, generator=gen) / math.sqrt(2)
|
||||
rr = (hr[:, :1] * s1 + hr[:, 1:2] * s2) / math.sqrt(2) + nr
|
||||
ri = (hi[:, :1] * s1 + hi[:, 1:2] * s2) / math.sqrt(2) + ni
|
||||
outs = []
|
||||
for u, (Mu, Qu) in enumerate(((M1, Q1), (M2, Q2))):
|
||||
hu_r = hr[:, u:u+1] / math.sqrt(2)
|
||||
hu_i = hi[:, u:u+1] / math.sqrt(2)
|
||||
mag = hu_r**2 + hu_i**2 + EPS_EQ
|
||||
xr = (rr @ Mu)
|
||||
xi = (ri @ Mu)
|
||||
xu = (xr * hu_r + xi * hu_i) / mag # Re(h* r'/(|h|^2+eps))
|
||||
sc = (xu @ Qu.T) / math.sqrt(D)
|
||||
outs.append(D * torch.softmax(sc, dim=1) * xu)
|
||||
gram = ((M1.T @ M1 - eye)**2).mean() \
|
||||
+ ((M2.T @ M2 - eye)**2).mean() \
|
||||
+ ((M1.T @ M2 - beta0 * eye)**2).mean()
|
||||
mse = ((outs[0] - e1)**2).mean() + ((outs[1] - e2)**2).mean()
|
||||
cs = torch.nn.functional.cosine_similarity(
|
||||
outs[0], e1, dim=1).mean() \
|
||||
+ torch.nn.functional.cosine_similarity(
|
||||
outs[1], e2, dim=1).mean()
|
||||
loss = l1 * gram + l2 * mse + l3 * (2.0 - cs)
|
||||
opt.zero_grad(); loss.backward()
|
||||
torch.nn.utils.clip_grad_norm_([M1, M2, Q1, Q2], 1.0)
|
||||
opt.step()
|
||||
if (ep + 1) % 50 == 0:
|
||||
print(f" epoch {ep+1}: loss {float(loss.detach()):.4f}",
|
||||
flush=True)
|
||||
print(f" trained in {time.time()-t0:.0f}s, "
|
||||
f"{4*D*D/1e6:.2f}M parameters")
|
||||
return (M1.detach().numpy(), M2.detach().numpy(),
|
||||
Q1.detach().numpy(), Q2.detach().numpy())
|
||||
|
||||
|
||||
def att_apply(model, r, h1, h2):
|
||||
M1, M2, Q1, Q2 = model
|
||||
outs = []
|
||||
for u, (Mu, Qu, hu) in enumerate(((M1, Q1, h1), (M2, Q2, h2))):
|
||||
xu = np.real(np.conj(hu) * (Mu.T @ r)) / (abs(hu)**2 + EPS_EQ)
|
||||
sc = (Qu @ xu) / math.sqrt(D)
|
||||
sc = sc - sc.max()
|
||||
w = np.exp(sc); w /= w.sum()
|
||||
outs.append(D * w * xu)
|
||||
return outs
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# ToDMA-adapted on real embeddings (complex channel)
|
||||
# ------------------------------------------------------------------
|
||||
def todma_prepare(V=1024, T=16):
|
||||
L = D // T
|
||||
Dict = rng.standard_normal((V, D))
|
||||
Dict /= np.linalg.norm(Dict, axis=1, keepdims=True)
|
||||
Sig = rng.standard_normal((V, L))
|
||||
Sig /= np.linalg.norm(Sig, axis=1, keepdims=True)
|
||||
amp = math.sqrt(1.0 / T) # E_b = 1 per user per block
|
||||
return Dict, Sig, amp, T, L
|
||||
|
||||
|
||||
def omp_code(Dict, e, T):
|
||||
resid = e.copy(); idx = []
|
||||
for _ in range(T):
|
||||
corr = np.abs(Dict @ resid)
|
||||
if idx:
|
||||
corr[idx] = -1
|
||||
k = int(corr.argmax()); idx.append(k)
|
||||
A = Dict[idx].T
|
||||
coef, *_ = np.linalg.lstsq(A, e, rcond=None)
|
||||
resid = e - A @ coef
|
||||
return idx, coef
|
||||
|
||||
|
||||
def todma_run(tod, codes, h, sig, noise_slots):
|
||||
Dict, Sig, amp, T, L = tod
|
||||
U = len(codes)
|
||||
det = [set() for _ in range(U)]
|
||||
for t in range(T):
|
||||
y = sum(h[u] * amp * Sig[codes[u][0][t]] for u in range(U)) \
|
||||
+ sig * noise_slots[t]
|
||||
resid = y.copy(); support = []
|
||||
for _ in range(U):
|
||||
corr = np.abs(Sig @ resid.conj())
|
||||
if support:
|
||||
corr[support] = -1
|
||||
kk = int(corr.argmax()); support.append(kk)
|
||||
Ah = (Sig[support].T * amp).astype(complex)
|
||||
coef, *_ = np.linalg.lstsq(Ah, y, rcond=None)
|
||||
resid = y - Ah @ coef
|
||||
sset = set(support)
|
||||
for u in range(U):
|
||||
if codes[u][0][t] in sset:
|
||||
det[u].add(codes[u][0][t])
|
||||
recs = []
|
||||
for u in range(U):
|
||||
idx, coef = codes[u]
|
||||
keep = [i for i, tid in enumerate(idx) if tid in det[u]]
|
||||
recs.append(sum(coef[i] * Dict[idx[i]] for i in keep)
|
||||
if keep else None)
|
||||
return recs
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
def main():
|
||||
A, B, betas = load_pairs()
|
||||
npairs = len(A)
|
||||
model = train_attention(float(betas.mean()))
|
||||
tod = todma_prepare()
|
||||
codes = [(omp_code(tod[0], A[i], tod[3]),
|
||||
omp_code(tod[0], B[i], tod[3])) for i in range(npairs)]
|
||||
print("[todma] sparse codes prepared")
|
||||
|
||||
keys = ("edma", "oma", "genie", "att", "att_x", "todma")
|
||||
res = {k: np.zeros(len(SNRS)) for k in keys}
|
||||
cnt = {k: np.zeros(len(SNRS)) for k in keys}
|
||||
t0 = time.time()
|
||||
for i in range(npairs):
|
||||
e1, e2, bi = A[i], B[i], float(betas[i])
|
||||
gi = 1.0 - bi**2
|
||||
c1, c2 = codes[i]
|
||||
for f in range(NFADE):
|
||||
U1, U2 = haar(D), haar(D)
|
||||
M1 = U1
|
||||
M2 = bi * U1 + math.sqrt(gi) * U2
|
||||
h = (rng.standard_normal(2) + 1j * rng.standard_normal(2)) \
|
||||
/ math.sqrt(2)
|
||||
h1, h2 = h
|
||||
r0 = h1 * (M1 @ e1) + h2 * (M2 @ e2)
|
||||
n = (rng.standard_normal(D) + 1j * rng.standard_normal(D)) \
|
||||
/ math.sqrt(2)
|
||||
n2 = (rng.standard_normal(D) + 1j * rng.standard_normal(D)) \
|
||||
/ math.sqrt(2)
|
||||
nslots = [(rng.standard_normal(tod[4])
|
||||
+ 1j * rng.standard_normal(tod[4])) / math.sqrt(2)
|
||||
for _ in range(tod[3])]
|
||||
# attention scheme transmits with ITS OWN trained masks
|
||||
r0a = h1 * (model[0] @ e1) + h2 * (model[1] @ e2)
|
||||
for k, s in enumerate(SNRS):
|
||||
sig = 10 ** (-s / 20.0)
|
||||
r = r0 + sig * n
|
||||
t1 = M1.T @ r / h1; t2 = M2.T @ r / h2
|
||||
g1 = (t1 - bi * (h2 / h1) * t2) / gi
|
||||
g2 = (t2 - bi * (h1 / h2) * t1) / gi
|
||||
res["edma"][k] += 0.5 * (cosine(g1, e1) + cosine(g2, e2))
|
||||
o1 = e1 + math.sqrt(2) * sig * n / h1
|
||||
o2 = e2 + math.sqrt(2) * sig * n2 / h2
|
||||
res["oma"][k] += 0.5 * (cosine(o1, e1) + cosine(o2, e2))
|
||||
ge1 = M1.T @ (r - h2 * (M2 @ e2)) / h1
|
||||
ge2 = M2.T @ (r - h1 * (M1 @ e1)) / h2
|
||||
res["genie"][k] += 0.5 * (cosine(ge1, e1) + cosine(ge2, e2))
|
||||
a1, a2 = att_apply(model, r0a + sig * n, h1, h2)
|
||||
res["att"][k] += 0.5 * (cosine(a1, e1) + cosine(a2, e2))
|
||||
res["att_x"][k] += 0.5 * (cosine(a1, e2) + cosine(a2, e1))
|
||||
for kk in ("edma", "oma", "genie", "att", "att_x"):
|
||||
cnt[kk][k] += 1
|
||||
if f < 40: # ToDMA heavier: 40 fading draws
|
||||
recs = todma_run(tod, (c1, c2), (h1, h2), sig, nslots)
|
||||
got = [cosine(recs[j], (e1, e2)[j])
|
||||
for j in range(2) if recs[j] is not None]
|
||||
if got:
|
||||
res["todma"][k] += float(np.mean(got))
|
||||
cnt["todma"][k] += 1
|
||||
print(f" pair {i+1}/{npairs} done ({time.time()-t0:.0f}s)",
|
||||
flush=True)
|
||||
for k in keys:
|
||||
res[k] /= np.maximum(cnt[k], 1)
|
||||
|
||||
fig, ax = plt.subplots()
|
||||
ax.plot(SNRS, res["edma"], "o-", color="C3", label="EDMA (closed form)")
|
||||
ax.plot(SNRS, res["att"], "s--", color="C0",
|
||||
label="Attention-based (retrained)")
|
||||
ax.plot(SNRS, res["todma"], "d-.", color="C4", label="ToDMA-adapted")
|
||||
ax.plot(SNRS, res["oma"], "v:", color="C1", label="OMA")
|
||||
ax.plot(SNRS, res["genie"], "-", color="gray", lw=1.0,
|
||||
label="Genie-aided SIC bound")
|
||||
ax.set_xlabel("Per-block SNR $\\rho$ [dB]")
|
||||
ax.set_ylabel("Mean cosine similarity")
|
||||
ax.set_xlim(SNRS[0], SNRS[-1]); ax.set_ylim(0, 0.85)
|
||||
ax.legend(loc="upper left")
|
||||
fig.subplots_adjust(**AXES_RECT)
|
||||
fig.savefig(FIG / "fig_bertvit_merged.pdf")
|
||||
plt.close(fig)
|
||||
print(f"[OK] wrote {FIG/'fig_bertvit_merged.pdf'}")
|
||||
|
||||
with open(DATA / "bertvit_merged.csv", "w", newline="") as fcsv:
|
||||
w = csv.writer(fcsv)
|
||||
w.writerow(["snr_db"] + list(keys))
|
||||
for k, s in enumerate(SNRS):
|
||||
w.writerow([s] + [res[key][k] for key in keys])
|
||||
print(f"[OK] wrote {DATA/'bertvit_merged.csv'}")
|
||||
for k, s in enumerate(SNRS):
|
||||
print(f" {s:4.0f} dB EDMA {res['edma'][k]:.3f} "
|
||||
f"ATT {res['att'][k]:.3f} (x {res['att_x'][k]:.3f}) "
|
||||
f"ToDMA {res['todma'][k]:.3f} OMA {res['oma'][k]:.3f} "
|
||||
f"genie {res['genie'][k]:.3f}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,187 @@
|
||||
"""
|
||||
Capacity-matched EDMA refinement (parameter budget equal to the
|
||||
attention scheme: 4 d^2 = 2.36M at d = 768).
|
||||
================================================================
|
||||
Four-head averaged gated refinement applied to the closed-form
|
||||
demultiplexer output:
|
||||
|
||||
out = (1/4) sum_k D softmax(Q_k x / sqrt(D)) .* x,
|
||||
|
||||
with Q_1..Q_4 in R^{D x D} (4 d^2 parameters, exactly the
|
||||
attention scheme's budget). The single-gate 0.59M refiner is the
|
||||
special case of four identical heads, so the family contains it
|
||||
by construction. Same training recipe: demux outputs from
|
||||
parametric pairs at beta = 0.028, Haar pool 32, Rayleigh
|
||||
channels, complex noise, training SNR uniform in [5, 25] dB,
|
||||
Adam 5e-4 with gradient clipping, batch 48, 200 epochs.
|
||||
|
||||
Evaluation on the real BERT/ViT pairs with fresh Haar masks and
|
||||
200 fading draws per pair. Appends column `edma_ref2` to
|
||||
data/bertvit_merged.csv and prints all-curve numbers.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import csv
|
||||
import math
|
||||
import time
|
||||
import numpy as np
|
||||
import torch
|
||||
|
||||
from fig_real_merged import load_pairs, cosine, SNRS, NFADE, D, DATA
|
||||
|
||||
SEED = 2026
|
||||
rng = np.random.default_rng(SEED + 31)
|
||||
torch.manual_seed(SEED + 31)
|
||||
BETA0 = 0.028
|
||||
G0 = 1.0 - BETA0**2
|
||||
|
||||
|
||||
def haar_t(gen):
|
||||
Q, R = torch.linalg.qr(torch.randn(D, D, generator=gen))
|
||||
return Q * torch.sign(torch.diagonal(R))
|
||||
|
||||
|
||||
def train_refiner2(epochs=220, steps=20, batch=48, lr=5e-4,
|
||||
l2=0.5, l3=0.5, pool=32):
|
||||
"""Stage 1 trains a single gate (the proven 0.59M recipe); stage 2
|
||||
warm-starts four heads from it plus small perturbations and
|
||||
fine-tunes at a reduced learning rate, so the capacity-matched
|
||||
family starts at the single-gate solution it contains."""
|
||||
print(f"=== training capacity-matched refinement (4-head gate, "
|
||||
f"4d^2 = {4*D*D/1e6:.2f}M params, warm-started) ===",
|
||||
flush=True)
|
||||
gen = torch.Generator().manual_seed(SEED + 31)
|
||||
masks = []
|
||||
for _ in range(pool):
|
||||
U1, U2 = haar_t(gen), haar_t(gen)
|
||||
masks.append((U1.numpy(), (BETA0 * U1
|
||||
+ math.sqrt(G0) * U2).numpy()))
|
||||
Q0 = torch.nn.Parameter(torch.randn(D, D, generator=gen)
|
||||
/ math.sqrt(D))
|
||||
params = [Q0]
|
||||
opt = torch.optim.Adam(params, lr=lr)
|
||||
stage2_at = 120 # epochs of single-gate pre-training
|
||||
|
||||
def forward(x):
|
||||
outs = [D * torch.softmax((x @ Qk.T) / math.sqrt(D), dim=1) * x
|
||||
for Qk in params]
|
||||
return sum(outs) / len(params)
|
||||
|
||||
t0 = time.time()
|
||||
for ep in range(epochs):
|
||||
if ep == stage2_at:
|
||||
base = params[0].detach()
|
||||
params = [torch.nn.Parameter(
|
||||
base.clone() + 0.02 * torch.randn(D, D, generator=gen)
|
||||
/ math.sqrt(D)) for _ in range(4)]
|
||||
opt = torch.optim.Adam(params, lr=2e-4)
|
||||
print(f" [warm start] 4 heads initialised from the trained "
|
||||
f"gate at epoch {ep}", flush=True)
|
||||
for _ in range(steps):
|
||||
xs, ts = [], []
|
||||
for _ in range(batch):
|
||||
e1 = torch.nn.functional.normalize(
|
||||
torch.randn(D, generator=gen), dim=0).numpy()
|
||||
w = torch.randn(D, generator=gen).numpy()
|
||||
w = w - (w @ e1) * e1
|
||||
w = w / np.linalg.norm(w)
|
||||
e2 = BETA0 * e1 + math.sqrt(G0) * w
|
||||
M1, M2 = masks[int(torch.randint(pool, (1,),
|
||||
generator=gen))]
|
||||
snr = float(5.0 + 20.0 * torch.rand(1, generator=gen))
|
||||
sig = 10 ** (-snr / 20.0)
|
||||
h = (torch.randn(2, generator=gen).numpy()
|
||||
+ 1j * torch.randn(2, generator=gen).numpy()) \
|
||||
/ math.sqrt(2)
|
||||
nc = (torch.randn(D, generator=gen).numpy()
|
||||
+ 1j * torch.randn(D, generator=gen).numpy()) \
|
||||
/ math.sqrt(2)
|
||||
rc = h[0] * (M1 @ e1) + h[1] * (M2 @ e2) + sig * nc
|
||||
t1 = M1.T @ rc / h[0]
|
||||
t2 = M2.T @ rc / h[1]
|
||||
g1 = (t1 - BETA0 * (h[1] / h[0]) * t2) / G0
|
||||
xs.append(torch.tensor(np.real(g1), dtype=torch.float32))
|
||||
ts.append(torch.tensor(e1, dtype=torch.float32))
|
||||
x = torch.stack(xs); t = torch.stack(ts)
|
||||
out = forward(x)
|
||||
mse = ((out - t)**2).mean()
|
||||
cs = torch.nn.functional.cosine_similarity(out, t, dim=1).mean()
|
||||
loss = l2 * mse + l3 * (1.0 - cs)
|
||||
opt.zero_grad(); loss.backward()
|
||||
torch.nn.utils.clip_grad_norm_(params, 1.0)
|
||||
opt.step()
|
||||
if (ep + 1) % 50 == 0:
|
||||
print(f" epoch {ep+1}: loss {float(loss.detach()):.4f} "
|
||||
f"(cos {float(cs.detach()):.3f})", flush=True)
|
||||
print(f" trained in {time.time()-t0:.0f}s")
|
||||
return [p.detach().numpy() for p in params]
|
||||
|
||||
|
||||
def refine2(P, g):
|
||||
x = np.real(g)
|
||||
|
||||
def gate(Q, v):
|
||||
sc = (Q @ v) / math.sqrt(D)
|
||||
sc = sc - sc.max()
|
||||
w = np.exp(sc); w /= w.sum()
|
||||
return D * w * v
|
||||
|
||||
return sum(gate(Qk, x) for Qk in P) / 4.0
|
||||
|
||||
|
||||
def main():
|
||||
A, B, betas = load_pairs()
|
||||
P = train_refiner2()
|
||||
ref = np.zeros(len(SNRS)); cnt = 0
|
||||
t0 = time.time()
|
||||
for i in range(len(A)):
|
||||
e1, e2, bi = A[i], B[i], float(betas[i])
|
||||
gi = 1.0 - bi**2
|
||||
for f in range(NFADE):
|
||||
G1 = rng.standard_normal((D, D))
|
||||
Qh, Rh = np.linalg.qr(G1)
|
||||
U1 = Qh * np.sign(np.diag(Rh))
|
||||
G2 = rng.standard_normal((D, D))
|
||||
Qh, Rh = np.linalg.qr(G2)
|
||||
U2 = Qh * np.sign(np.diag(Rh))
|
||||
M1 = U1
|
||||
M2 = bi * U1 + math.sqrt(gi) * U2
|
||||
h = (rng.standard_normal(2) + 1j * rng.standard_normal(2)) \
|
||||
/ math.sqrt(2)
|
||||
h1, h2 = h
|
||||
r0 = h1 * (M1 @ e1) + h2 * (M2 @ e2)
|
||||
n = (rng.standard_normal(D) + 1j * rng.standard_normal(D)) \
|
||||
/ math.sqrt(2)
|
||||
for k, s in enumerate(SNRS):
|
||||
sig = 10 ** (-s / 20.0)
|
||||
r = r0 + sig * n
|
||||
t1 = M1.T @ r / h1; t2 = M2.T @ r / h2
|
||||
g1 = (t1 - bi * (h2 / h1) * t2) / gi
|
||||
g2 = (t2 - bi * (h1 / h2) * t1) / gi
|
||||
ref[k] += 0.5 * (cosine(refine2(P, g1), e1)
|
||||
+ cosine(refine2(P, g2), e2))
|
||||
cnt += 1
|
||||
print(f" pair {i+1}/{len(A)} done ({time.time()-t0:.0f}s)",
|
||||
flush=True)
|
||||
ref /= cnt
|
||||
|
||||
rows = list(csv.DictReader(open(DATA / "bertvit_merged.csv")))
|
||||
names = list(rows[0].keys())
|
||||
if "edma_ref2" not in names:
|
||||
names.append("edma_ref2")
|
||||
for k, r in enumerate(rows):
|
||||
r["edma_ref2"] = f"{ref[k]}"
|
||||
with open(DATA / "bertvit_merged.csv", "w", newline="") as f:
|
||||
w = csv.DictWriter(f, fieldnames=names)
|
||||
w.writeheader(); w.writerows(rows)
|
||||
print("[OK] appended edma_ref2 to bertvit_merged.csv")
|
||||
for k, r in enumerate(rows):
|
||||
print(f" {float(r['snr_db']):4.0f} dB "
|
||||
f"EDMA {float(r['edma']):.3f} "
|
||||
f"ref(0.59M) {float(r['edma_ref']):.3f} "
|
||||
f"ref2(2.36M) {ref[k]:.3f} "
|
||||
f"ATT(2.36M) {float(r['att']):.3f} "
|
||||
f"genie {float(r['genie']):.3f}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,41 @@
|
||||
"""Canonical replot of fig_bertvit_merged.pdf from data/bertvit_merged.csv.
|
||||
Curves: EDMA, EDMA + refinement (hybrid), ToDMA-adapted, OMA, genie bound.
|
||||
The attention columns remain in the CSV but are not plotted."""
|
||||
import csv
|
||||
from pathlib import Path
|
||||
import matplotlib
|
||||
matplotlib.use("Agg")
|
||||
import matplotlib.pyplot as plt
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
plt.rcParams.update({
|
||||
"font.family": "serif",
|
||||
"font.serif": ["DejaVu Serif", "Times New Roman"],
|
||||
"font.size": 9, "axes.labelsize": 9, "legend.fontsize": 6.6,
|
||||
"xtick.labelsize": 8, "ytick.labelsize": 8,
|
||||
"axes.grid": True, "grid.linestyle": "--", "grid.linewidth": 0.4,
|
||||
"grid.alpha": 0.6, "lines.linewidth": 1.4, "lines.markersize": 4.0,
|
||||
"figure.figsize": (3.15, 2.36), "pdf.fonttype": 42,
|
||||
})
|
||||
AXES_RECT = dict(left=0.205, right=0.965, top=0.955, bottom=0.185)
|
||||
|
||||
rows = list(csv.DictReader(open(ROOT / "data" / "bertvit_merged.csv")))
|
||||
snr = [float(r["snr_db"]) for r in rows]
|
||||
col = lambda k: [float(r[k]) for r in rows]
|
||||
|
||||
fig, ax = plt.subplots()
|
||||
ax.plot(snr, col("edma"), "o-", color="C3", label="EDMA (closed form)")
|
||||
ax.plot(snr, col("edma_ref"), "^-", color="C2",
|
||||
label="EDMA + refinement stage")
|
||||
ax.plot(snr, col("todma"), "d-.", color="C4", label="ToDMA-adapted")
|
||||
ax.plot(snr, col("oma"), "v:", color="C1", label="OMA")
|
||||
ax.plot(snr, col("genie"), "-", color="gray", lw=1.0,
|
||||
label="Genie-aided SIC bound")
|
||||
ax.set_xlabel("Per-block SNR $\\rho$ [dB]")
|
||||
ax.set_ylabel("Mean cosine similarity")
|
||||
ax.set_xlim(snr[0], snr[-1])
|
||||
ax.set_ylim(0, 0.85)
|
||||
ax.legend(loc="upper left")
|
||||
fig.subplots_adjust(**AXES_RECT)
|
||||
fig.savefig(ROOT / "fig" / "fig_bertvit_merged.pdf")
|
||||
print("[OK] wrote fig_bertvit_merged.pdf (no attention curve)")
|
||||
@@ -0,0 +1,43 @@
|
||||
"""Canonical replot of fig_sic.pdf from data/sic_comparison.csv
|
||||
(realizable analog SIC vs genie SIC vs EDMA vs OMA, beta = 0.311,
|
||||
d = 512, block-Rayleigh). US-spelling labels, uniform geometry."""
|
||||
import csv
|
||||
from pathlib import Path
|
||||
import matplotlib
|
||||
matplotlib.use("Agg")
|
||||
import matplotlib.pyplot as plt
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
plt.rcParams.update({
|
||||
"font.family": "serif",
|
||||
"font.serif": ["DejaVu Serif", "Times New Roman"],
|
||||
"font.size": 9, "axes.labelsize": 9, "legend.fontsize": 6.6,
|
||||
"xtick.labelsize": 8, "ytick.labelsize": 8,
|
||||
"axes.grid": True, "grid.linestyle": "--", "grid.linewidth": 0.4,
|
||||
"grid.alpha": 0.6, "lines.linewidth": 1.4, "lines.markersize": 4.0,
|
||||
"figure.figsize": (3.15, 2.36), "pdf.fonttype": 42,
|
||||
})
|
||||
AXES_RECT = dict(left=0.205, right=0.965, top=0.955, bottom=0.185)
|
||||
|
||||
rows = list(csv.DictReader(open(ROOT / "data" / "sic_comparison.csv")))
|
||||
snr = [float(r["snr_db"]) for r in rows]
|
||||
col = lambda k: [float(r[k]) for r in rows]
|
||||
|
||||
fig, ax = plt.subplots()
|
||||
ax.plot(snr, col("edma"), "o-", color="C3", label="EDMA (closed form)")
|
||||
ax.plot(snr, col("sic"), "^-.", color="C2", label="Realizable analog SIC")
|
||||
ax.plot(snr, col("oma"), "v:", color="C1", label="OMA")
|
||||
ax.plot(snr, col("genie"), "-", color="gray", lw=1.0,
|
||||
label="Genie-aided SIC bound")
|
||||
ax.set_xlabel("Per-block SNR $\\rho$ [dB]")
|
||||
ax.set_ylabel("Mean cosine similarity")
|
||||
ax.set_xlim(snr[0], snr[-1])
|
||||
ax.set_ylim(0, 0.7)
|
||||
ax.legend(loc="upper left")
|
||||
fig.subplots_adjust(**AXES_RECT)
|
||||
fig.savefig(ROOT / "fig" / "fig_sic.pdf")
|
||||
print("[OK] wrote fig_sic.pdf")
|
||||
for r in rows:
|
||||
print(f" {float(r['snr_db']):4.0f} dB EDMA {float(r['edma']):.3f} "
|
||||
f"SIC {float(r['sic']):.3f} genie {float(r['genie']):.3f} "
|
||||
f"OMA {float(r['oma']):.3f}")
|
||||
@@ -0,0 +1,617 @@
|
||||
"""
|
||||
Revision simulations for the EDMA TCOM resubmission.
|
||||
=======================================================
|
||||
Implements the per-realisation (finite-d) analysis and the corrected
|
||||
energy-normalised rate accounting, plus the reviewer-requested
|
||||
experiments:
|
||||
|
||||
E0 Theorem-1 verification: exact self-interference constant C_SI
|
||||
E1 fig_floor : per-user MSE vs block SNR, interference floor
|
||||
E2 fig_sic : realisable SIC vs genie SIC vs EDMA vs OMA
|
||||
E3 (text numbers) : Rayleigh unconditional MSE, ZF vs regularised
|
||||
E4 fig_csi : imperfect-CSI robustness
|
||||
E5 fig_maskfam : Walsh-Hadamard structured masks vs Haar
|
||||
E6 fig_coop : high-affinity combining-mode crossover
|
||||
E7 fig_rate_corrected, fig_beta_sweep_corrected, fig_multiuser_corrected
|
||||
|
||||
Conventions (identical to the revised manuscript):
|
||||
* unit per-block transmit energy E_b = 1 per user
|
||||
* rho = E_b / sigma_n^2 (per-block received SNR; per-symbol SNR rho/d)
|
||||
* block-Rayleigh h ~ CN(0,1) unless the AWGN point |h|=1 is stated
|
||||
* complex AWGN CN(0, sigma^2 I_d); embeddings real, unit norm
|
||||
* orientation convention <e1,e2> = +beta
|
||||
Fixed seed. CSVs -> ../fig, PDFs -> ../fig_toc.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import csv
|
||||
import math
|
||||
from pathlib import Path
|
||||
import numpy as np
|
||||
import matplotlib
|
||||
matplotlib.use("Agg")
|
||||
import matplotlib.pyplot as plt
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
CSV_DIR = ROOT / "data"; CSV_DIR.mkdir(exist_ok=True)
|
||||
FIG_DIR = ROOT / "fig"; FIG_DIR.mkdir(exist_ok=True)
|
||||
|
||||
plt.rcParams.update({
|
||||
"font.family": "serif",
|
||||
"font.serif": ["DejaVu Serif", "Times New Roman"],
|
||||
"font.size": 9, "axes.labelsize": 9, "axes.titlesize": 9,
|
||||
"legend.fontsize": 7.0, "xtick.labelsize": 8, "ytick.labelsize": 8,
|
||||
"axes.grid": True, "grid.linestyle": "--", "grid.linewidth": 0.4,
|
||||
"grid.alpha": 0.6, "lines.linewidth": 1.4, "lines.markersize": 4.0,
|
||||
"figure.figsize": (3.15, 2.36), "pdf.fonttype": 42,
|
||||
})
|
||||
AXES_RECT = dict(left=0.205, right=0.965, top=0.955, bottom=0.185)
|
||||
|
||||
rng = np.random.default_rng(2026)
|
||||
|
||||
|
||||
def save_fig(fig, name):
|
||||
p = FIG_DIR / f"{name}.pdf"
|
||||
fig.subplots_adjust(**AXES_RECT)
|
||||
fig.savefig(p)
|
||||
plt.close(fig)
|
||||
print(f"[OK] wrote {p}")
|
||||
|
||||
|
||||
def write_csv(name, header, rows):
|
||||
p = CSV_DIR / f"{name}.csv"
|
||||
with open(p, "w", newline="") as f:
|
||||
w = csv.writer(f); w.writerow(header); w.writerows(rows)
|
||||
print(f"[OK] wrote {p}")
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Core constructions
|
||||
# ------------------------------------------------------------------
|
||||
def haar(d):
|
||||
G = rng.standard_normal((d, d))
|
||||
Q, R = np.linalg.qr(G)
|
||||
return Q * np.sign(np.diag(R))
|
||||
|
||||
|
||||
def unit(v):
|
||||
return v / np.linalg.norm(v)
|
||||
|
||||
|
||||
def embed_pair(d, beta):
|
||||
"""e1, e2 real unit vectors with <e1,e2> = +beta."""
|
||||
e1 = unit(rng.standard_normal(d))
|
||||
w = rng.standard_normal(d)
|
||||
w = unit(w - (w @ e1) * e1)
|
||||
e2 = beta * e1 + math.sqrt(1.0 - beta**2) * w
|
||||
return e1, e2
|
||||
|
||||
|
||||
def two_user_masks(d, beta, U1=None, U2=None):
|
||||
if U1 is None: U1 = haar(d)
|
||||
if U2 is None: U2 = haar(d)
|
||||
g = math.sqrt(1.0 - beta**2)
|
||||
return U1, beta * U1 + g * U2
|
||||
|
||||
|
||||
def rayleigh(n=1):
|
||||
return (rng.standard_normal(n) + 1j * rng.standard_normal(n)) / math.sqrt(2)
|
||||
|
||||
|
||||
def C_SI(beta, c):
|
||||
"""User-1 self-interference constant (exact to O(1/d)), <e1,e2>=+beta."""
|
||||
g = 1.0 - beta**2
|
||||
return (g**2 * abs(c)**2 + beta**2 + beta**4 * abs(c)**2
|
||||
+ 2.0 * beta**4 * np.real(c)) / g
|
||||
|
||||
|
||||
def C_SI2(beta, c2):
|
||||
"""User-2 self-interference constant (deterministic), c2 = h1/h2."""
|
||||
g = 1.0 - beta**2
|
||||
return (abs(c2)**2 + beta**2 + 2.0 * beta**2 * np.real(c2)) / g
|
||||
|
||||
|
||||
def C_bar(beta):
|
||||
"""Symmetrised constant at |h|=1 (block-alternating mask roles)."""
|
||||
return 0.5 * (C_SI(beta, 1.0 + 0j) + C_SI2(beta, 1.0 + 0j))
|
||||
|
||||
|
||||
def demux(r, M1, M2, h1, h2, beta):
|
||||
"""beta-aware demultiplexer (13); returns (e1_hat, e2_hat)."""
|
||||
g = 1.0 - beta**2
|
||||
t1 = (M1.T @ r) / h1
|
||||
t2 = (M2.T @ r) / h2
|
||||
e1 = (t1 - beta * (h2 / h1) * t2) / g
|
||||
e2 = (t2 - beta * (h1 / h2) * t1) / g
|
||||
return e1, e2
|
||||
|
||||
|
||||
def cosine(a, b):
|
||||
return abs(np.vdot(a, b)) / (np.linalg.norm(a) * np.linalg.norm(b))
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# E0 : Theorem-1 verification
|
||||
# ------------------------------------------------------------------
|
||||
def E0_theorem_check(d=512, betas=(0.0, 0.311, 0.5, 0.7), ntr=300):
|
||||
print("\n=== E0: Theorem 1 (self-interference constant) verification ===")
|
||||
rows = []
|
||||
worst = 0.0
|
||||
for beta in betas:
|
||||
# random unit-modulus channels (AWGN-type magnitude, random phase)
|
||||
errs1, errs2 = [], []
|
||||
for _ in range(ntr):
|
||||
h1 = np.exp(1j * rng.uniform(0, 2 * np.pi))
|
||||
h2 = np.exp(1j * rng.uniform(0, 2 * np.pi))
|
||||
e1, e2 = embed_pair(d, beta)
|
||||
M1, M2 = two_user_masks(d, beta)
|
||||
r = h1 * (M1 @ e1) + h2 * (M2 @ e2) # noise-free
|
||||
g1, g2 = demux(r, M1, M2, h1, h2, beta)
|
||||
errs1.append(np.linalg.norm(g1 - e1)**2 / C_SI(beta, h2 / h1))
|
||||
errs2.append(np.linalg.norm(g2 - e2)**2 / C_SI2(beta, h1 / h2))
|
||||
r1, r2 = float(np.mean(errs1)), float(np.mean(errs2))
|
||||
dev = max(abs(r1 - 1.0), abs(r2 - 1.0)) * 100
|
||||
worst = max(worst, dev)
|
||||
print(f" beta={beta:.3f} MC/theory user1 = {r1:.4f}, user2 = {r2:.4f}"
|
||||
f" (max dev {dev:.2f}%)")
|
||||
rows.append([beta, r1, r2, dev])
|
||||
write_csv("theorem_check", ["beta", "user1_mc_over_theory",
|
||||
"user2_mc_over_theory", "max_dev_pct"], rows)
|
||||
print(f" worst-case deviation {worst:.2f}%")
|
||||
return worst
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# E1 : interference floor (MSE vs block SNR), AWGN point |h|=1
|
||||
# ------------------------------------------------------------------
|
||||
def E1_floor(beta=0.311, dims=(256, 768), snr_db=np.arange(0, 41, 2.5), ntr=150):
|
||||
print("\n=== E1: finite-d interference floor ===")
|
||||
g = 1.0 - beta**2
|
||||
csi = C_SI(beta, 1.0 + 0j)
|
||||
fig, ax = plt.subplots()
|
||||
colors = {256: "C0", 768: "C3"}
|
||||
rows = []
|
||||
for d in dims:
|
||||
mc = np.zeros(len(snr_db))
|
||||
for _ in range(ntr):
|
||||
e1, e2 = embed_pair(d, beta)
|
||||
M1, M2 = two_user_masks(d, beta)
|
||||
r0 = (M1 @ e1) + (M2 @ e2)
|
||||
n = (rng.standard_normal(d) + 1j * rng.standard_normal(d)) / math.sqrt(2)
|
||||
for k, s in enumerate(snr_db):
|
||||
sig = 10 ** (-s / 20.0)
|
||||
g1, _ = demux(r0 + sig * n, M1, M2, 1.0, 1.0, beta)
|
||||
mc[k] += np.linalg.norm(g1 - e1)**2
|
||||
mc /= ntr
|
||||
rho = 10 ** (snr_db / 10.0)
|
||||
th = d / (rho * g) + csi
|
||||
ideal = d / (rho * g)
|
||||
ax.semilogy(snr_db, mc, "o", ms=3.5, color=colors[d], mfc="none",
|
||||
label=rf"MC, $d={d}$")
|
||||
ax.semilogy(snr_db, th, "-", color=colors[d],
|
||||
label=rf"Theorem 1, $d={d}$")
|
||||
if d == dims[-1]:
|
||||
ax.semilogy(snr_db, ideal, ":", color="k", lw=1.1,
|
||||
label="Idealized (no floor)")
|
||||
for s, m, t, i in zip(snr_db, mc, th, ideal):
|
||||
rows.append([d, s, m, t, i])
|
||||
onset = 10 * math.log10(d / (g * csi))
|
||||
print(f" d={d}: floor C_SI={csi:.4f}, onset ~{onset:.1f} dB, "
|
||||
f"max MC/theory dev "
|
||||
f"{100*max(abs(mc/th-1)):.1f}%")
|
||||
ax.axhline(csi, color="gray", lw=0.8, ls="--")
|
||||
ax.text(1.0, csi * 1.15, r"floor $C_{\mathrm{SI}}$", fontsize=7, color="gray")
|
||||
ax.set_xlabel("Per-block SNR $\\rho$ [dB]")
|
||||
ax.set_ylabel(r"Per-user MSE $\mathbb{E}\|\hat{\mathbf{e}}_u-\mathbf{e}_u\|_2^2$")
|
||||
ax.set_xlim(0, 40); ax.set_ylim(0.5, 2000)
|
||||
ax.legend(loc="upper right", ncol=1)
|
||||
save_fig(fig, "fig_floor")
|
||||
write_csv("floor_validation", ["d", "snr_db", "mse_mc", "mse_theory", "mse_ideal"], rows)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# E2 : realisable SIC vs genie SIC vs EDMA vs OMA (Rayleigh)
|
||||
# ------------------------------------------------------------------
|
||||
def E2_sic(beta=0.311, d=512, snr_db=np.arange(0, 31, 5), ntr=400):
|
||||
print("\n=== E2: realisable vs genie SIC (Rayleigh) ===")
|
||||
res = {k: np.zeros(len(snr_db)) for k in
|
||||
("edma", "oma", "genie", "sic")}
|
||||
for _ in range(ntr):
|
||||
e1, e2 = embed_pair(d, beta)
|
||||
M1, M2 = two_user_masks(d, beta)
|
||||
h1, h2 = rayleigh(2)
|
||||
r0 = h1 * (M1 @ e1) + h2 * (M2 @ e2)
|
||||
n = (rng.standard_normal(d) + 1j * rng.standard_normal(d)) / math.sqrt(2)
|
||||
n2 = (rng.standard_normal(d) + 1j * rng.standard_normal(d)) / math.sqrt(2)
|
||||
for k, s in enumerate(snr_db):
|
||||
sig = 10 ** (-s / 20.0)
|
||||
r = r0 + sig * n
|
||||
# EDMA
|
||||
g1, g2 = demux(r, M1, M2, h1, h2, beta)
|
||||
res["edma"][k] += 0.5 * (cosine(g1, e1) + cosine(g2, e2))
|
||||
# OMA equivalent-bandwidth model: interference-free, noise x sqrt(2)
|
||||
o1 = e1 + math.sqrt(2) * sig * n / h1
|
||||
o2 = e2 + math.sqrt(2) * sig * n2 / h2
|
||||
res["oma"][k] += 0.5 * (cosine(o1, e1) + cosine(o2, e2))
|
||||
# genie SIC: perfect removal of the other user for BOTH users
|
||||
ge1 = M1.T @ (r - h2 * (M2 @ e2)) / h1
|
||||
ge2 = M2.T @ (r - h1 * (M1 @ e1)) / h2
|
||||
res["genie"][k] += 0.5 * (cosine(ge1, e1) + cosine(ge2, e2))
|
||||
# realisable SIC: stronger user first (matched filter),
|
||||
# unit-norm projection as the analog decision, then subtract
|
||||
if abs(h1) >= abs(h2):
|
||||
hs, hw, Ms, Mw, es, ew = h1, h2, M1, M2, e1, e2
|
||||
else:
|
||||
hs, hw, Ms, Mw, es, ew = h2, h1, M2, M1, e2, e1
|
||||
d_s = Ms.T @ r / hs
|
||||
dec_s = d_s / np.linalg.norm(d_s) # analog decision
|
||||
r_res = r - hs * (Ms @ dec_s)
|
||||
d_w = Mw.T @ r_res / hw
|
||||
res["sic"][k] += 0.5 * (cosine(d_s, es) + cosine(d_w, ew))
|
||||
for k in res:
|
||||
res[k] /= ntr
|
||||
fig, ax = plt.subplots()
|
||||
ax.plot(snr_db, res["edma"], "o-", color="C3", label="EDMA")
|
||||
ax.plot(snr_db, res["genie"], "s--", color="C0", label="Genie-aided SIC")
|
||||
ax.plot(snr_db, res["sic"], "^-.", color="C2", label="Realisable SIC")
|
||||
ax.plot(snr_db, res["oma"], "v:", color="C1", label="OMA")
|
||||
ax.set_xlabel("Per-block SNR $\\rho$ [dB]")
|
||||
ax.set_ylabel("Mean cosine similarity")
|
||||
ax.set_xlim(snr_db[0], snr_db[-1]); ax.set_ylim(0, 1)
|
||||
ax.legend(loc="upper left")
|
||||
save_fig(fig, "fig_sic")
|
||||
rows = [[s] + [res[k][i] for k in ("edma", "oma", "genie", "sic")]
|
||||
for i, s in enumerate(snr_db)]
|
||||
write_csv("sic_comparison", ["snr_db", "edma", "oma", "genie", "sic"], rows)
|
||||
i20 = list(snr_db).index(20)
|
||||
print(f" at 20 dB: EDMA {res['edma'][i20]:.3f}, realisable SIC "
|
||||
f"{res['sic'][i20]:.3f}, genie {res['genie'][i20]:.3f}, "
|
||||
f"OMA {res['oma'][i20]:.3f}")
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# E3 : Rayleigh unconditional MSE ??ZF inversion vs regularised
|
||||
# ------------------------------------------------------------------
|
||||
def E3_regularised(beta=0.311, d=512, snrs=(10, 20), ntr=4000):
|
||||
print("\n=== E3: Rayleigh unconditional MSE, ZF vs regularised ===")
|
||||
rows = []
|
||||
for s in snrs:
|
||||
sig = 10 ** (-s / 20.0)
|
||||
sig2 = sig**2
|
||||
mse_zf, mse_rg = [], []
|
||||
for _ in range(ntr):
|
||||
e1, e2 = embed_pair(d, beta)
|
||||
M1, M2 = two_user_masks(d, beta)
|
||||
h1, h2 = rayleigh(2)
|
||||
r = h1 * (M1 @ e1) + h2 * (M2 @ e2) \
|
||||
+ sig * (rng.standard_normal(d) + 1j * rng.standard_normal(d)) / math.sqrt(2)
|
||||
g1, _ = demux(r, M1, M2, h1, h2, beta)
|
||||
mse_zf.append(np.linalg.norm(g1 - e1)**2)
|
||||
# regularised inversion: 1/h -> h*/(|h|^2 + d sigma^2)
|
||||
eps = d * sig2
|
||||
f1 = (abs(h1)**2 + eps) / np.conj(h1)
|
||||
f2 = (abs(h2)**2 + eps) / np.conj(h2)
|
||||
g1r, _ = demux(r, M1, M2, f1, f2, beta)
|
||||
mse_rg.append(np.linalg.norm(g1r - e1)**2)
|
||||
zf_mean, zf_med = float(np.mean(mse_zf)), float(np.median(mse_zf))
|
||||
rg_mean, rg_med = float(np.mean(mse_rg)), float(np.median(mse_rg))
|
||||
print(f" {s} dB: ZF mean {zf_mean:9.2f} (median {zf_med:6.2f}) | "
|
||||
f"regularised mean {rg_mean:6.3f} (median {rg_med:6.3f})")
|
||||
rows.append([s, zf_mean, zf_med, rg_mean, rg_med])
|
||||
write_csv("rayleigh_mse", ["snr_db", "zf_mean", "zf_median",
|
||||
"reg_mean", "reg_median"], rows)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# E4 : imperfect CSI
|
||||
# ------------------------------------------------------------------
|
||||
def E4_csi(beta=0.311, d=512, snr=30.0,
|
||||
sh2=np.array([0.0, 0.01, 0.02, 0.05, 0.1, 0.2, 0.3]), ntr=400):
|
||||
"""EDMA cosine is CSI-direction-invariant (h-estimates cancel in the
|
||||
demux direction); realisable SIC degrades through its subtraction stage."""
|
||||
print("\n=== E4: imperfect CSI robustness (EDMA vs realisable SIC) ===")
|
||||
sig = 10 ** (-snr / 20.0)
|
||||
res_e = np.zeros(len(sh2)); res_s = np.zeros(len(sh2))
|
||||
for _ in range(ntr):
|
||||
e1, e2 = embed_pair(d, beta)
|
||||
M1, M2 = two_user_masks(d, beta)
|
||||
h1, h2 = rayleigh(2)
|
||||
r = h1 * (M1 @ e1) + h2 * (M2 @ e2) + sig * (
|
||||
rng.standard_normal(d) + 1j * rng.standard_normal(d)) / math.sqrt(2)
|
||||
eps1, eps2 = rayleigh(2)
|
||||
for j, v in enumerate(sh2):
|
||||
hh1 = h1 + math.sqrt(v) * eps1
|
||||
hh2 = h2 + math.sqrt(v) * eps2
|
||||
g1, g2 = demux(r, M1, M2, hh1, hh2, beta)
|
||||
res_e[j] += 0.5 * (cosine(g1, e1) + cosine(g2, e2))
|
||||
# realisable SIC with the same imperfect estimates
|
||||
if abs(hh1) >= abs(hh2):
|
||||
hs, hw, Ms, Mw, es, ew = hh1, hh2, M1, M2, e1, e2
|
||||
else:
|
||||
hs, hw, Ms, Mw, es, ew = hh2, hh1, M2, M1, e2, e1
|
||||
d_s = Ms.T @ r / hs
|
||||
dec_s = d_s / np.linalg.norm(d_s)
|
||||
r_res = r - hs * (Ms @ dec_s)
|
||||
d_w = Mw.T @ r_res / hw
|
||||
res_s[j] += 0.5 * (cosine(d_s, es) + cosine(d_w, ew))
|
||||
res_e /= ntr; res_s /= ntr
|
||||
print(f" EDMA: {res_e[0]:.4f} -> {res_e[-1]:.4f} "
|
||||
f"(delta {100*(res_e[0]-res_e[-1]):.2f} points)")
|
||||
print(f" SIC : {res_s[0]:.4f} -> {res_s[-1]:.4f} "
|
||||
f"(delta {100*(res_s[0]-res_s[-1]):.2f} points)")
|
||||
fig, ax = plt.subplots()
|
||||
ax.plot(sh2, res_e, "o-", color="C3", label="EDMA")
|
||||
ax.plot(sh2, res_s, "^-.", color="C2", label="Realisable SIC")
|
||||
ax.set_xlabel(r"CSI error variance $\sigma_h^2$")
|
||||
ax.set_ylabel("Mean cosine similarity")
|
||||
ax.set_xlim(0, sh2[-1]); ax.set_ylim(0, 0.7)
|
||||
ax.legend(loc="lower left")
|
||||
save_fig(fig, "fig_csi")
|
||||
rows = [[v, res_e[j], res_s[j]] for j, v in enumerate(sh2)]
|
||||
write_csv("csi_error", ["sigma_h2", "edma", "sic"], rows)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# E5 : Walsh-Hadamard structured masks vs Haar
|
||||
# ------------------------------------------------------------------
|
||||
def hadamard(n):
|
||||
H = np.array([[1.0]])
|
||||
while H.shape[0] < n:
|
||||
H = np.block([[H, H], [H, -H]])
|
||||
return H / math.sqrt(n)
|
||||
|
||||
|
||||
def E5_maskfam(beta=0.311, d=512, snr_db=np.arange(0, 41, 5), ntr=200):
|
||||
print("\n=== E5: Walsh-Hadamard masks vs Haar mixture ===")
|
||||
H = hadamard(d)
|
||||
g = math.sqrt(1.0 - beta**2)
|
||||
res = {"haar": np.zeros(len(snr_db)), "wh": np.zeros(len(snr_db))}
|
||||
for _ in range(ntr):
|
||||
e1, e2 = embed_pair(d, beta)
|
||||
M1, M2 = two_user_masks(d, beta)
|
||||
D1 = np.diag(rng.choice([-1.0, 1.0], d))
|
||||
D2 = np.diag(rng.choice([-1.0, 1.0], d))
|
||||
W1 = H @ D1
|
||||
W2 = beta * W1 + g * (H @ D2)
|
||||
r0h = (M1 @ e1) + (M2 @ e2)
|
||||
r0w = (W1 @ e1) + (W2 @ e2)
|
||||
n = (rng.standard_normal(d) + 1j * rng.standard_normal(d)) / math.sqrt(2)
|
||||
for k, s in enumerate(snr_db):
|
||||
sig = 10 ** (-s / 20.0)
|
||||
g1, _ = demux(r0h + sig * n, M1, M2, 1.0, 1.0, beta)
|
||||
w1, _ = demux(r0w + sig * n, W1, W2, 1.0, 1.0, beta)
|
||||
res["haar"][k] += cosine(g1, e1)
|
||||
res["wh"][k] += cosine(w1, e1)
|
||||
for k in res:
|
||||
res[k] /= ntr
|
||||
dev = 100 * np.max(np.abs(res["wh"] - res["haar"]))
|
||||
print(f" max |WH - Haar| cosine deviation: {dev:.2f} points")
|
||||
fig, ax = plt.subplots()
|
||||
ax.plot(snr_db, res["haar"], "o-", color="C3",
|
||||
label=r"Haar mixture, $\mathcal{O}(d^2)$")
|
||||
ax.plot(snr_db, res["wh"], "s--", color="C0",
|
||||
label=r"Walsh-Hadamard, $\mathcal{O}(d\log d)$")
|
||||
ax.set_xlabel("Per-block SNR $\\rho$ [dB]")
|
||||
ax.set_ylabel("Mean cosine similarity")
|
||||
ax.set_xlim(snr_db[0], snr_db[-1]); ax.set_ylim(0, 0.8)
|
||||
ax.legend(loc="upper left")
|
||||
save_fig(fig, "fig_maskfam")
|
||||
rows = [[s, res["haar"][i], res["wh"][i]] for i, s in enumerate(snr_db)]
|
||||
write_csv("mask_family_rev", ["snr_db", "haar", "wh"], rows)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# E6 : high-affinity combining mode
|
||||
# ------------------------------------------------------------------
|
||||
def E6_coop(d=512, snr=20.0, betas=np.linspace(0.0, 0.98, 21), ntr=100):
|
||||
print("\n=== E6: high-affinity combining-mode crossover ===")
|
||||
sig = 10 ** (-snr / 20.0)
|
||||
pairs = [(haar(d), haar(d)) for _ in range(ntr)]
|
||||
chans = [rayleigh(2) for _ in range(ntr)]
|
||||
cos_dx = np.zeros(len(betas)); cos_cb = np.zeros(len(betas))
|
||||
for j, beta in enumerate(betas):
|
||||
for t in range(ntr):
|
||||
U1, U2 = pairs[t]
|
||||
h1, h2 = chans[t]
|
||||
e1, e2 = embed_pair(d, beta)
|
||||
M1, M2 = two_user_masks(d, beta, U1, U2)
|
||||
r = h1 * (M1 @ e1) + h2 * (M2 @ e2) + sig * (
|
||||
rng.standard_normal(d) + 1j * rng.standard_normal(d)) / math.sqrt(2)
|
||||
g1, _ = demux(r, M1, M2, h1, h2, beta)
|
||||
cos_dx[j] += cosine(g1, e1)
|
||||
# affinity combining: coherent weights for the e1 component
|
||||
a1 = h1 + beta**2 * h2
|
||||
a2 = beta * (h1 + h2)
|
||||
comb = np.conj(a1) * (M1.T @ r) + np.conj(a2) * (M2.T @ r)
|
||||
cos_cb[j] += cosine(comb, e1)
|
||||
cos_dx /= ntr; cos_cb /= ntr
|
||||
ix = np.where(cos_cb >= cos_dx)[0]
|
||||
cross = betas[ix[0]] if len(ix) else float("nan")
|
||||
print(f" crossover affinity ~ {cross:.2f} at rho={snr:.0f} dB")
|
||||
fig, ax = plt.subplots()
|
||||
ax.plot(betas, cos_dx, "o-", color="C3", label="Separation mode (demux)")
|
||||
ax.plot(betas, cos_cb, "s--", color="C0", label="Combining mode")
|
||||
ax.set_xlabel(r"Pairwise affinity $\beta$")
|
||||
ax.set_ylabel("Mean cosine similarity")
|
||||
ax.set_xlim(0, 1); ax.set_ylim(0, 0.8)
|
||||
ax.legend(loc="lower left")
|
||||
save_fig(fig, "fig_coop")
|
||||
rows = [[b, cos_dx[i], cos_cb[i]] for i, b in enumerate(betas)]
|
||||
write_csv("coop_mode", ["beta", "cos_demux", "cos_combine"], rows)
|
||||
return cross
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# E7 : corrected effective-rate figures
|
||||
# ------------------------------------------------------------------
|
||||
def eta_edma(rho, d, beta, csi=None):
|
||||
g = 1.0 - beta**2
|
||||
if csi is None:
|
||||
csi = C_bar(beta) # symmetrised constant (alternating masks)
|
||||
return 1.0 / (d / (rho * g) + csi)
|
||||
|
||||
|
||||
def E7_rates(beta=0.311, d=512):
|
||||
print("\n=== E7a: corrected effective-rate comparison ===")
|
||||
snr_db = np.arange(0, 31, 1.0)
|
||||
rho = 10 ** (snr_db / 10.0)
|
||||
g = 1.0 - beta**2
|
||||
T_edma = 2 * np.log2(1 + eta_edma(rho, d, beta))
|
||||
T_ideal = 2 * np.log2(1 + rho * g / d)
|
||||
T_oma = 2 * np.log2(1 + rho / (2 * d))
|
||||
T_genie = 2 * np.log2(1 + rho / d)
|
||||
C_mac = np.log2(1 + 2 * rho / d)
|
||||
fig, ax = plt.subplots()
|
||||
ax.plot(snr_db, T_edma, "-", color="C3", label="EDMA (Theorem 1)")
|
||||
ax.plot(snr_db, T_ideal, ":", color="C3", lw=1.1,
|
||||
label="EDMA idealized (infeasible)")
|
||||
ax.plot(snr_db, T_oma, "--", color="C1", label="OMA")
|
||||
ax.plot(snr_db, T_genie, "-.", color="C0", label="Genie-aided SIC bound")
|
||||
ax.plot(snr_db, C_mac, "-", color="k", lw=1.0, label="MAC sum capacity")
|
||||
ax.set_xlabel("Per-block SNR $\\rho$ [dB]")
|
||||
ax.set_ylabel("Effective sum rate [bps/Hz]")
|
||||
ax.set_xlim(0, 30); ax.set_ylim(0, 3.2)
|
||||
ax.legend(loc="upper left")
|
||||
save_fig(fig, "fig_rate_corrected")
|
||||
rows = [[s, T_edma[i], T_ideal[i], T_oma[i], T_genie[i], C_mac[i]]
|
||||
for i, s in enumerate(snr_db)]
|
||||
write_csv("rate_corrected",
|
||||
["snr_db", "edma", "edma_ideal", "oma", "genie", "mac"], rows)
|
||||
i20 = list(snr_db).index(20.0)
|
||||
csi = C_bar(beta)
|
||||
rho_c = d * (2 - 1 / g) / csi
|
||||
print(f" at 20 dB: EDMA {T_edma[i20]:.3f}, OMA {T_oma[i20]:.3f} "
|
||||
f"(gain {T_edma[i20]/T_oma[i20]:.2f}x), MAC {C_mac[i20]:.3f}, "
|
||||
f"EDMA/MAC {T_edma[i20]/C_mac[i20]:.3f} (gamma={g:.3f})")
|
||||
print(f" OMA re-crossover rho_c = {10*math.log10(rho_c):.1f} dB")
|
||||
|
||||
print("\n=== E7b: corrected beta sweep ===")
|
||||
betas = np.linspace(0.0, 0.98, 99)
|
||||
fig, ax = plt.subplots()
|
||||
rows = []
|
||||
for s, col in ((10, "C0"), (20, "C3")):
|
||||
rho_s = 10 ** (s / 10.0)
|
||||
Te = np.array([2 * np.log2(1 + eta_edma(rho_s, d, b)) for b in betas])
|
||||
To = 2 * np.log2(1 + rho_s / (2 * d))
|
||||
Tg = 2 * np.log2(1 + rho_s / d)
|
||||
ax.plot(betas, Te, "-", color=col, label=rf"EDMA, $\rho={s}$ dB")
|
||||
ax.axhline(To, color=col, ls="--", lw=1.0,
|
||||
label=rf"OMA, $\rho={s}$ dB")
|
||||
ax.axhline(Tg, color=col, ls="-.", lw=0.8,
|
||||
label=rf"Genie-aided SIC, $\rho={s}$ dB")
|
||||
ix = np.where(Te <= To)[0]
|
||||
bstar = betas[ix[0]] if len(ix) else float("nan")
|
||||
print(f" rho={s} dB: crossover beta* = {bstar:.3f} "
|
||||
f"(wideband limit 1/sqrt(2)=0.707)")
|
||||
for i, b in enumerate(betas):
|
||||
rows.append([s, b, Te[i], To, Tg])
|
||||
for b0 in (0.031, 0.311):
|
||||
ax.axvline(b0, color="gray", ls=":", lw=0.9)
|
||||
ax.set_xlabel(r"Pairwise affinity $\beta$")
|
||||
ax.set_ylabel("Effective sum rate [bps/Hz]")
|
||||
ax.set_xlim(0, 1); ax.set_ylim(0, 1.02)
|
||||
ax.set_yticks([0.0, 0.2, 0.4, 0.6, 0.8, 1.0])
|
||||
ax.legend(loc="upper right", ncol=1, fontsize=5.8,
|
||||
handlelength=1.5, borderaxespad=0.2)
|
||||
save_fig(fig, "fig_beta_sweep_corrected")
|
||||
write_csv("beta_sweep_corrected",
|
||||
["snr_db", "beta", "edma", "oma", "genie"], rows)
|
||||
|
||||
|
||||
def E7_multiuser(beta=0.311, d=512, Us=(2, 3, 4), ntr_cal=80, ntr_mc=120):
|
||||
print("\n=== E7c: corrected multi-user scaling ===")
|
||||
snr_db = np.arange(0, 31, 2.5)
|
||||
snr_mk = np.arange(0, 31, 5)
|
||||
rho = 10 ** (snr_db / 10.0)
|
||||
fig, ax = plt.subplots()
|
||||
colors = {2: "C0", 3: "C2", 4: "C3"}
|
||||
rows = []
|
||||
csi2 = C_SI(beta, 1.0 + 0j)
|
||||
for U in Us:
|
||||
B = (1 - beta) * np.eye(U) + beta * np.ones((U, U))
|
||||
Binv_uu = np.linalg.inv(B)[0, 0]
|
||||
gU = 1.0 / Binv_uu
|
||||
# calibrate C_SI^(U) by noise-free MC at h_u = 1 (the same
|
||||
# evaluation convention as the two-user rate curves, so the
|
||||
# U = 2 curve reduces exactly to T_EDMA with C_bar),
|
||||
# averaged over all users (mask roles are asymmetric)
|
||||
acc = 0.0
|
||||
for _ in range(ntr_cal):
|
||||
A = np.linalg.cholesky(B)
|
||||
Uks = [haar(d) for _ in range(U)]
|
||||
Ms = [sum(A[u, k] * Uks[k] for k in range(U)) for u in range(U)]
|
||||
h = np.ones(U, dtype=complex)
|
||||
# symmetric equal-affinity embeddings: e_u = beta-mixed set
|
||||
base = unit(rng.standard_normal(d))
|
||||
es = []
|
||||
for u in range(U):
|
||||
w = rng.standard_normal(d)
|
||||
w = unit(w - (w @ base) * base)
|
||||
# construct so that <e_u,e_v> ~ beta pairwise
|
||||
es.append(unit(math.sqrt(beta) * base
|
||||
+ math.sqrt(1 - beta) * w))
|
||||
r = sum(h[u] * (Ms[u] @ es[u]) for u in range(U))
|
||||
Binv = np.linalg.inv(B)
|
||||
# block demux e_hat_u = (1/h_u) sum_v Binv[u,v] M_v^T r
|
||||
for u in range(U):
|
||||
eh = sum(Binv[u, v] * (Ms[v].T @ r) for v in range(U)) / h[u]
|
||||
acc += np.linalg.norm(eh - es[u])**2
|
||||
csiU = acc / (ntr_cal * U)
|
||||
print(f" U={U}: C_SI^(U) = {csiU:.3f} "
|
||||
f"((U-1)*C_bar = {(U-1)*C_bar(beta):.3f}), gamma_U = {gU:.3f}")
|
||||
eta = 1.0 / (d * Binv_uu / rho + csiU)
|
||||
T_th = U * np.log2(1 + eta)
|
||||
T_oma = U * np.log2(1 + rho / (U * d))
|
||||
ax.plot(snr_db, T_th, "-", color=colors[U], label=rf"EDMA, $U={U}$")
|
||||
ax.plot(snr_db, T_oma, "--", color=colors[U], lw=1.0,
|
||||
label=rf"OMA, $U={U}$")
|
||||
# MC markers (with noise, h_u = 1, per-realization real masks)
|
||||
err_mc = np.zeros(len(snr_mk))
|
||||
for _ in range(ntr_mc):
|
||||
A = np.linalg.cholesky(B)
|
||||
Uks = [haar(d) for _ in range(U)]
|
||||
Ms = [sum(A[u, k] * Uks[k] for k in range(U)) for u in range(U)]
|
||||
h = np.ones(U, dtype=complex)
|
||||
base = unit(rng.standard_normal(d))
|
||||
es = []
|
||||
for u in range(U):
|
||||
w = rng.standard_normal(d)
|
||||
w = unit(w - (w @ base) * base)
|
||||
es.append(unit(math.sqrt(beta) * base
|
||||
+ math.sqrt(1 - beta) * w))
|
||||
r0 = sum(h[u] * (Ms[u] @ es[u]) for u in range(U))
|
||||
n = (rng.standard_normal(d) + 1j * rng.standard_normal(d)) / math.sqrt(2)
|
||||
Binv = np.linalg.inv(B)
|
||||
for k, s in enumerate(snr_mk):
|
||||
sig = 10 ** (-s / 20.0)
|
||||
r = r0 + sig * n
|
||||
for u in range(U):
|
||||
eh = sum(Binv[u, v] * (Ms[v].T @ r) for v in range(U)) / h[u]
|
||||
err_mc[k] += np.linalg.norm(eh - es[u])**2
|
||||
err_mc /= ntr_mc * U
|
||||
T_mc = U * np.log2(1 + 1.0 / err_mc)
|
||||
ax.plot(snr_mk, T_mc, "o", color=colors[U], ms=4, mfc="none")
|
||||
for i, s in enumerate(snr_db):
|
||||
rows.append([U, s, T_th[i], T_oma[i]])
|
||||
i20 = list(snr_db).index(20.0)
|
||||
print(f" at 20 dB: EDMA {T_th[i20]:.3f} vs OMA {T_oma[i20]:.3f} "
|
||||
f"(gain {T_th[i20]/T_oma[i20]:.2f}x)")
|
||||
ax.set_xlabel("Per-block SNR $\\rho$ [dB]")
|
||||
ax.set_ylabel("Effective sum rate [bps/Hz]")
|
||||
ax.set_xlim(0, 30); ax.set_ylim(0, 1.5)
|
||||
ax.legend(loc="upper left", ncol=1, fontsize=6.2)
|
||||
save_fig(fig, "fig_multiuser_corrected")
|
||||
write_csv("multiuser_corrected", ["U", "snr_db", "edma", "oma"], rows)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
todo = set(sys.argv[1:])
|
||||
ALL = {
|
||||
"E0": E0_theorem_check, "E1": E1_floor, "E2": E2_sic,
|
||||
"E3": E3_regularised, "E4": E4_csi, "E5": E5_maskfam,
|
||||
"E6": E6_coop, "E7a": E7_rates, "E7c": E7_multiuser,
|
||||
}
|
||||
for name, fn in ALL.items():
|
||||
if not todo or name in todo:
|
||||
fn()
|
||||
print("\nAll requested revision simulations complete.")
|
||||
@@ -0,0 +1,374 @@
|
||||
"""
|
||||
Complete numerical verification of every closed form in the manuscript.
|
||||
=======================================================================
|
||||
Each check implements the formula EXACTLY as printed in main.tex and
|
||||
compares it against a direct Monte-Carlo or algebraic evaluation.
|
||||
Prints PASS/FAIL per item with the achieved deviation. Fixed seed.
|
||||
|
||||
V1 per-realization Gram identity M1^T M2 = beta I + sqrt(g) Q
|
||||
V2 Theorem 1 full MSE (noise + C_SI,u) vs MC, random complex h
|
||||
V3 noise-free calibration of C_SI,1 / C_SI,2 (several phases)
|
||||
V4 quoted constants: C_SI,1, C_SI,2, C-bar at (0.311, h=1);
|
||||
cosine ceiling 1/sqrt(1+C_SI,1) = 0.70; rho_f = 28 dB at d=768
|
||||
V5 SINR corollary eta_u = 1/MSE (per-coordinate accounting)
|
||||
V6 C_SI,u >= 1 for all beta (proof identities gamma*C_SI,1 =
|
||||
gamma + 4 beta^4, gamma*C_SI,2 = 1 + 3 beta^2 at h=1)
|
||||
V7 Proposition (MAC consistency) on a (beta, rho) grid
|
||||
V8 wideband limit T/C_MAC -> gamma
|
||||
V9 beta* crossover roots at 10/20 dB (0.700 / 0.590, d=512)
|
||||
V10 rho_c = d(2-1/gamma)/C-bar exact iff-condition + 25.7 dB value
|
||||
V11 idealized no-floor variant crosses C_MAC at 2 beta^2 d/gamma^2
|
||||
(~21 dB at d=512, beta=0.311)
|
||||
V12 mismatch identity (eq:mismatch) + bound value 8.8e-3
|
||||
V13 CSI-direction invariance: |cos| unchanged under wrong h-hat;
|
||||
eq:csi-free equals eq:correct
|
||||
V14 cross-moment lemma E[n^H M_u M_v^T n] = sigma^2 beta d
|
||||
V15 multi-user [B^-1]_uu Sherman-Morrison formula, U = 2..6
|
||||
V16 multi-user noise-free C_SI^(U) ~ (U-1) C-bar (within 10 %)
|
||||
V17 Walsh-Hadamard masks: exact orthogonality + expected cross-Gram
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import math
|
||||
import numpy as np
|
||||
|
||||
def hadamard(n):
|
||||
H = np.array([[1.0]])
|
||||
while H.shape[0] < n:
|
||||
H = np.block([[H, H], [H, -H]])
|
||||
return H
|
||||
|
||||
|
||||
def brentq(f, a, b, tol=1e-12):
|
||||
fa, fb = f(a), f(b)
|
||||
assert fa * fb < 0, "no sign change"
|
||||
for _ in range(200):
|
||||
m = 0.5 * (a + b)
|
||||
fm = f(m)
|
||||
if abs(fm) < tol or (b - a) < tol:
|
||||
return m
|
||||
if fa * fm < 0:
|
||||
b, fb = m, fm
|
||||
else:
|
||||
a, fa = m, fm
|
||||
return 0.5 * (a + b)
|
||||
|
||||
rng = np.random.default_rng(2026)
|
||||
FAIL = []
|
||||
|
||||
|
||||
def report(name, ok, detail):
|
||||
tag = "PASS" if ok else "FAIL"
|
||||
if not ok:
|
||||
FAIL.append(name)
|
||||
print(f"[{tag}] {name}: {detail}")
|
||||
|
||||
|
||||
def haar(d):
|
||||
Q, R = np.linalg.qr(rng.standard_normal((d, d)))
|
||||
return Q * np.sign(np.diag(R))
|
||||
|
||||
|
||||
def unit(v):
|
||||
return v / np.linalg.norm(v)
|
||||
|
||||
|
||||
def pair(d, beta):
|
||||
e1 = unit(rng.standard_normal(d))
|
||||
w = rng.standard_normal(d)
|
||||
w = unit(w - (w @ e1) * e1)
|
||||
return e1, beta * e1 + math.sqrt(1 - beta**2) * w
|
||||
|
||||
|
||||
def csi1(beta, c):
|
||||
g = 1 - beta**2
|
||||
n2 = 1 + beta**2 * abs(c)**2 + 2 * beta**2 * np.real(c)
|
||||
return (g**2 * abs(c)**2 + beta**2 * n2) / g
|
||||
|
||||
|
||||
def csi2(beta, c):
|
||||
g = 1 - beta**2
|
||||
return (abs(c)**2 + beta**2 + 2 * beta**2 * np.real(c)) / g
|
||||
|
||||
|
||||
# ---------------- V1: per-realization Gram identity ----------------
|
||||
d, beta = 256, 0.311
|
||||
g = 1 - beta**2
|
||||
U1, U2 = haar(d), haar(d)
|
||||
M1, M2 = U1, beta * U1 + math.sqrt(g) * U2
|
||||
dev = np.abs(M1.T @ M2 - (beta * np.eye(d)
|
||||
+ math.sqrt(g) * U1.T @ U2)).max()
|
||||
report("V1 Gram identity", dev < 1e-12, f"max dev {dev:.2e}")
|
||||
|
||||
# ---------------- V2: Theorem 1 full MSE, random complex h ---------
|
||||
d = 512
|
||||
for beta in (0.1, 0.311, 0.5):
|
||||
g = 1 - beta**2
|
||||
h = (rng.standard_normal(2) + 1j * rng.standard_normal(2)) / math.sqrt(2)
|
||||
h1, h2 = h
|
||||
rho_db = 15.0
|
||||
sig = 10 ** (-rho_db / 20.0)
|
||||
e1, e2 = pair(d, beta)
|
||||
mc = np.zeros(2)
|
||||
NT = 300
|
||||
for _ in range(NT):
|
||||
U1, U2 = haar(d), haar(d)
|
||||
M1, M2 = U1, beta * U1 + math.sqrt(g) * U2
|
||||
n = (rng.standard_normal(d) + 1j * rng.standard_normal(d)) \
|
||||
/ math.sqrt(2)
|
||||
r = h1 * (M1 @ e1) + h2 * (M2 @ e2) + sig * n
|
||||
t1 = M1.T @ r / h1
|
||||
t2 = M2.T @ r / h2
|
||||
g1 = (t1 - beta * (h2 / h1) * t2) / g
|
||||
g2 = (t2 - beta * (h1 / h2) * t1) / g
|
||||
mc[0] += np.linalg.norm(g1 - e1)**2
|
||||
mc[1] += np.linalg.norm(g2 - e2)**2
|
||||
mc /= NT
|
||||
th1 = d * sig**2 / (abs(h1)**2 * g) + csi1(beta, h2 / h1)
|
||||
th2 = d * sig**2 / (abs(h2)**2 * g) + csi2(beta, h1 / h2)
|
||||
dev = max(abs(mc[0] / th1 - 1), abs(mc[1] / th2 - 1))
|
||||
report(f"V2 Theorem 1 MSE (beta={beta})", dev < 0.02,
|
||||
f"MC/theory dev {100*dev:.2f}% (O(1/d) at d={d})")
|
||||
|
||||
# ---------------- V3: noise-free C_SI calibration ------------------
|
||||
d = 512
|
||||
for phase in (0.0, math.pi / 3, math.pi):
|
||||
beta = 0.311
|
||||
g = 1 - beta**2
|
||||
h1 = 1.0 + 0j
|
||||
h2 = np.exp(1j * phase)
|
||||
e1, e2 = pair(d, beta)
|
||||
mc = np.zeros(2)
|
||||
NT = 200
|
||||
for _ in range(NT):
|
||||
U1, U2 = haar(d), haar(d)
|
||||
M1, M2 = U1, beta * U1 + math.sqrt(g) * U2
|
||||
r = h1 * (M1 @ e1) + h2 * (M2 @ e2)
|
||||
t1 = M1.T @ r / h1
|
||||
t2 = M2.T @ r / h2
|
||||
g1 = (t1 - beta * (h2 / h1) * t2) / g
|
||||
g2 = (t2 - beta * (h1 / h2) * t1) / g
|
||||
mc[0] += np.linalg.norm(g1 - e1)**2
|
||||
mc[1] += np.linalg.norm(g2 - e2)**2
|
||||
mc /= NT
|
||||
t1v, t2v = csi1(beta, h2 / h1), csi2(beta, h1 / h2)
|
||||
dev = max(abs(mc[0] / t1v - 1), abs(mc[1] / t2v - 1))
|
||||
report(f"V3 noise-free C_SI (phase={phase:.2f})", dev < 0.02,
|
||||
f"dev {100*dev:.2f}%")
|
||||
|
||||
# ---------------- V4: quoted constants -----------------------------
|
||||
beta = 0.311
|
||||
g = 1 - beta**2
|
||||
c1v, c2v = csi1(beta, 1.0 + 0j), csi2(beta, 1.0 + 0j)
|
||||
cbar = (c1v + c2v) / 2
|
||||
ceil1 = 1 / math.sqrt(1 + c1v)
|
||||
rho_f_db = 10 * math.log10(768 * g / c1v)
|
||||
ok = (abs(cbar - 1.2349) < 5e-4 and abs(ceil1 - 0.70) < 5e-3
|
||||
and abs(rho_f_db - 28) < 0.5)
|
||||
report("V4 quoted constants", ok,
|
||||
f"C_SI,1 {c1v:.4f}, C_SI,2 {c2v:.4f}, C-bar {cbar:.4f} "
|
||||
f"(quoted 1.2349), ceiling {ceil1:.4f} (quoted 0.70), "
|
||||
f"rho_f {rho_f_db:.1f} dB (quoted 28)")
|
||||
|
||||
# ---------------- V5: SINR = 1/MSE ---------------------------------
|
||||
rho = 10 ** (15 / 10)
|
||||
eta = 1 / (512 / (rho * g) + c1v)
|
||||
mse = 512 / (rho * g) + c1v
|
||||
report("V5 SINR corollary", abs(eta * mse - 1) < 1e-12,
|
||||
f"eta*MSE = {eta*mse:.6f}")
|
||||
|
||||
# ---------------- V6: C_SI >= 1 and proof identities ---------------
|
||||
ok = True
|
||||
worst = 1e9
|
||||
for b in np.linspace(0.0, 0.99, 200):
|
||||
gg = 1 - b**2
|
||||
lhs1 = gg * csi1(b, 1.0 + 0j)
|
||||
lhs2 = gg * csi2(b, 1.0 + 0j)
|
||||
if abs(lhs1 - (gg + 4 * b**4)) > 1e-12: ok = False
|
||||
if abs(lhs2 - (1 + 3 * b**2)) > 1e-12: ok = False
|
||||
worst = min(worst, csi1(b, 1.0 + 0j), csi2(b, 1.0 + 0j))
|
||||
report("V6 C_SI >= 1 + proof identities", ok and worst >= 1 - 1e-12,
|
||||
f"min C_SI over beta grid = {worst:.6f}")
|
||||
|
||||
# ---------------- V7: MAC consistency on a grid --------------------
|
||||
def T_edma(b, r_, d_):
|
||||
gg = 1 - b**2
|
||||
cb = (csi1(b, 1 + 0j) + csi2(b, 1 + 0j)) / 2
|
||||
return 2 * np.log2(1 + 1 / (d_ / (r_ * gg) + cb))
|
||||
|
||||
ok = True
|
||||
for b in np.linspace(0, 0.95, 40):
|
||||
for rdb in np.linspace(-10, 60, 60):
|
||||
r_ = 10 ** (rdb / 10)
|
||||
gg = 1 - b**2
|
||||
mid = np.log2(1 + 2 * r_ * gg / 512)
|
||||
cmac = np.log2(1 + 2 * r_ / 512)
|
||||
if T_edma(b, r_, 512) > mid + 1e-12 or mid > cmac + 1e-12:
|
||||
ok = False
|
||||
report("V7 MAC consistency grid", ok, "T_EDMA <= log2(1+2 rho g/d) <= C_MAC")
|
||||
|
||||
# ---------------- V8: wideband limit -------------------------------
|
||||
b = 0.311
|
||||
r_ = 1e-6 * 512
|
||||
lim = T_edma(b, r_, 512) / np.log2(1 + 2 * r_ / 512)
|
||||
report("V8 wideband limit", abs(lim - (1 - b**2)) < 1e-3,
|
||||
f"T/C_MAC at rho/d=1e-6: {lim:.5f} vs gamma {1-b**2:.5f}")
|
||||
|
||||
# ---------------- V9: beta* crossover roots ------------------------
|
||||
def beta_star(rdb, d_=512):
|
||||
r_ = 10 ** (rdb / 10)
|
||||
T_oma = 2 * np.log2(1 + r_ / (2 * d_))
|
||||
return brentq(lambda b: T_edma(b, r_, d_) - T_oma, 0.3, 0.9)
|
||||
|
||||
b10, b20 = beta_star(10), beta_star(20)
|
||||
report("V9 beta* crossover", abs(b10 - 0.700) < 5e-3
|
||||
and abs(b20 - 0.590) < 5e-3,
|
||||
f"10 dB: {b10:.3f} (quoted 0.700), 20 dB: {b20:.3f} (quoted 0.590)")
|
||||
|
||||
# ---------------- V10: rho_c iff-condition + value -----------------
|
||||
b = 0.311
|
||||
gg = 1 - b**2
|
||||
cb = (csi1(b, 1 + 0j) + csi2(b, 1 + 0j)) / 2
|
||||
rho_c = 512 * (2 - 1 / gg) / cb
|
||||
rho_c_db = 10 * math.log10(rho_c)
|
||||
eps = 1e-4
|
||||
below = T_edma(b, rho_c * (1 - eps), 512) \
|
||||
- 2 * np.log2(1 + rho_c * (1 - eps) / 1024)
|
||||
above = T_edma(b, rho_c * (1 + eps), 512) \
|
||||
- 2 * np.log2(1 + rho_c * (1 + eps) / 1024)
|
||||
report("V10 rho_c crossover", below > 0 > above
|
||||
and abs(rho_c_db - 25.7) < 0.1,
|
||||
f"rho_c {rho_c_db:.2f} dB (quoted 25.7), sign flip verified")
|
||||
|
||||
# ---------------- V11: idealized-MAC crossing ----------------------
|
||||
rho_x = 2 * b**2 * 512 / gg**2
|
||||
f = lambda r_: 2 * np.log2(1 + r_ * gg / 512) - np.log2(1 + 2 * r_ / 512)
|
||||
root = brentq(f, 10.0, 1e4)
|
||||
report("V11 idealized crossing", abs(root / rho_x - 1) < 1e-6
|
||||
and abs(10 * math.log10(root) - 21) < 0.3,
|
||||
f"root {10*math.log10(root):.2f} dB, formula 2b^2d/g^2 "
|
||||
f"{10*math.log10(rho_x):.2f} dB (quoted ~21)")
|
||||
|
||||
# ---------------- V12: mismatch identity + bound value -------------
|
||||
b, delta = 0.3, 0.06
|
||||
bh = b + delta
|
||||
hr = 1.0 + 0j
|
||||
e1, e2 = pair(64, b)
|
||||
t1 = e1 + b * hr * e2 # expected-Gram surrogate outputs
|
||||
t2v_ = e2 + b * np.conj(hr) * e1
|
||||
g1 = (t1 - bh * hr * t2v_) / (1 - bh**2)
|
||||
lhs = g1 - e1
|
||||
rhs = delta / (1 - bh**2) * (bh * e1 - hr * e2)
|
||||
dev = np.linalg.norm(lhs - rhs)
|
||||
bound = delta**2 * (abs(bh) + abs(hr))**2 / (1 - bh**2)**2
|
||||
report("V12 mismatch identity", dev < 1e-12
|
||||
and abs(bound - 8.8e-3) < 2e-4,
|
||||
f"identity dev {dev:.1e}, bound {bound:.4f} (quoted 8.8e-3)")
|
||||
|
||||
# ---------------- V13: CSI-direction invariance --------------------
|
||||
d = 256
|
||||
b = 0.311
|
||||
g = 1 - b**2
|
||||
e1, e2 = pair(d, b)
|
||||
U1, U2 = haar(d), haar(d)
|
||||
M1, M2 = U1, b * U1 + math.sqrt(g) * U2
|
||||
h1, h2 = 0.7 - 0.4j, -0.2 + 1.1j
|
||||
n = (rng.standard_normal(d) + 1j * rng.standard_normal(d)) / math.sqrt(2)
|
||||
r = h1 * (M1 @ e1) + h2 * (M2 @ e2) + 0.1 * n
|
||||
truec = (M1.T @ r / h1 - b * (h2 / h1) * (M2.T @ r / h2)) / g
|
||||
csif = (M1 - b * M2).T @ r / (h1 * g)
|
||||
dev1 = np.abs(truec - csif).max()
|
||||
h1w = h1 * (1.5 * np.exp(0.8j)) # badly wrong estimate
|
||||
wrong = (M1 - b * M2).T @ r / (h1w * g)
|
||||
c_true = abs(np.vdot(truec, e1)) / (np.linalg.norm(truec))
|
||||
c_wrong = abs(np.vdot(wrong, e1)) / (np.linalg.norm(wrong))
|
||||
report("V13 CSI invariance", dev1 < 1e-12 and abs(c_true - c_wrong) < 1e-12,
|
||||
f"csi-free identity dev {dev1:.1e}, |cos| unchanged "
|
||||
f"({c_true:.6f} vs {c_wrong:.6f})")
|
||||
|
||||
# ---------------- V14: cross-moment lemma --------------------------
|
||||
d = 256
|
||||
b = 0.311
|
||||
sig2 = 0.5
|
||||
acc = 0.0
|
||||
NT = 4000
|
||||
U1, U2 = haar(d), haar(d)
|
||||
M1, M2 = U1, b * U1 + math.sqrt(1 - b**2) * U2
|
||||
for _ in range(NT):
|
||||
n = math.sqrt(sig2) * (rng.standard_normal(d)
|
||||
+ 1j * rng.standard_normal(d)) / math.sqrt(2)
|
||||
acc += np.real(np.conj(n) @ (M1 @ (M2.T @ n)))
|
||||
acc /= NT
|
||||
th = sig2 * b * d
|
||||
report("V14 cross-moment lemma", abs(acc / th - 1) < 0.05,
|
||||
f"MC {acc:.3f} vs sigma^2 beta d {th:.3f} "
|
||||
f"({100*abs(acc/th-1):.1f}%)")
|
||||
|
||||
# ---------------- V15: [B^-1]_uu Sherman-Morrison ------------------
|
||||
ok = True
|
||||
for U in range(2, 7):
|
||||
for b in (0.1, 0.311, 0.6):
|
||||
B = (1 - b) * np.eye(U) + b * np.ones((U, U))
|
||||
num = 1 + (U - 2) * b
|
||||
den = (1 - b) * (1 + (U - 1) * b)
|
||||
if abs(np.linalg.inv(B)[0, 0] - num / den) > 1e-12:
|
||||
ok = False
|
||||
report("V15 [B^-1]_uu formula", ok, "U=2..6, beta grid, exact")
|
||||
|
||||
# ---------------- V16: multi-user C_SI^(U) -------------------------
|
||||
d = 512
|
||||
b = 0.311
|
||||
g = 1 - b**2
|
||||
cb = (csi1(b, 1 + 0j) + csi2(b, 1 + 0j)) / 2
|
||||
for U in (3, 4):
|
||||
B = (1 - b) * np.eye(U) + b * np.ones((U, U))
|
||||
Binv = np.linalg.inv(B)
|
||||
es = []
|
||||
e1 = unit(rng.standard_normal(d))
|
||||
for u in range(U):
|
||||
if u == 0:
|
||||
es.append(e1)
|
||||
else:
|
||||
w = rng.standard_normal(d)
|
||||
w = unit(w - (w @ e1) * e1)
|
||||
es.append(b * e1 + math.sqrt(g) * w)
|
||||
mse = 0.0
|
||||
NT = 60
|
||||
for _ in range(NT):
|
||||
Us = [haar(d) for _ in range(U)]
|
||||
Ms = [Us[0]]
|
||||
for u in range(1, U):
|
||||
Ms.append(b * Us[0] + math.sqrt(g) * Us[u])
|
||||
r = sum(Ms[u] @ es[u] for u in range(U)) # h_u = 1
|
||||
t = np.stack([Ms[u].T @ r for u in range(U)])
|
||||
rec = np.einsum("uv,vd->ud", Binv, t)
|
||||
mse += np.linalg.norm(rec[0] - es[0])**2
|
||||
mse /= NT
|
||||
ratio = mse / ((U - 1) * cb)
|
||||
report(f"V16 C_SI^(U) additivity (U={U})", abs(ratio - 1) < 0.10,
|
||||
f"noise-free MSE {mse:.3f} vs (U-1)C-bar "
|
||||
f"{(U-1)*cb:.3f} (ratio {ratio:.3f})")
|
||||
|
||||
# ---------------- V17: Walsh-Hadamard masks ------------------------
|
||||
d = 256
|
||||
H = hadamard(d) / math.sqrt(d)
|
||||
b = 0.311
|
||||
acc = np.zeros((d, d))
|
||||
NT = 400
|
||||
for _ in range(NT):
|
||||
D1 = np.diag(rng.choice([-1.0, 1.0], d))
|
||||
D2 = np.diag(rng.choice([-1.0, 1.0], d))
|
||||
W1 = H @ D1
|
||||
W2 = b * W1 + math.sqrt(1 - b**2) * H @ D2
|
||||
acc += W1.T @ W2 / NT
|
||||
orth = np.abs((H @ np.diag(rng.choice([-1.0, 1.0], d))).T
|
||||
@ (H @ np.diag(rng.choice([-1.0, 1.0], d)))
|
||||
@ np.ones(d) / d).max()
|
||||
diag_dev = abs(np.diag(acc).mean() - b)
|
||||
off = np.abs(acc - np.diag(np.diag(acc))).mean()
|
||||
report("V17 WH masks", diag_dev < 0.02 and off < 0.01,
|
||||
f"E[cross-Gram] diag {np.diag(acc).mean():.4f} vs beta {b}, "
|
||||
f"mean |off-diag| {off:.4f}")
|
||||
|
||||
print()
|
||||
print("=" * 60)
|
||||
print(f"RESULT: {'ALL PASS' if not FAIL else 'FAILURES: ' + ', '.join(FAIL)}")
|
||||
Reference in New Issue
Block a user