v2 design: independent masks + affinity-aware Wiener demultiplexer

Redesign after the independent-mask dominance finding: the affinity
now parameterizes the receiver (closed-form Wiener) instead of the
mask ensemble. New Theorem 1 (spectral closed form), floors
sqrt(1-b^2)/2 vs 1/2, full-cooperation bound with equality at b=1.
GPU (torch) Monte Carlo backend, decision-directed SIC baseline,
TikZ block diagram source, verification suite V1-V11.
This commit is contained in:
KiHoLee
2026-08-17 02:12:10 +09:00
parent b8b853e62e
commit 358faecc0c
29 changed files with 1777 additions and 1514 deletions
+99 -185
View File
@@ -1,30 +1,30 @@
"""
Merged real-data comparison figure (replaces separate Figs 4 and 5).
Real-data comparison on cached BERT (text) + ViT (image) pairs.
===================================================================
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:
Evaluates the schemes on the cached real 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).
Schemes (v2 design: independent Haar masks per user):
1. EDMA : affinity-aware Wiener demultiplexer with the
per-pair measured beta_i.
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
4. 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.
The hybrid (EDMA + refinement stage) curve is produced separately by
refine_matched.py (torch) and merged by replot_merged.py.
EDMA/OMA/genie run in torch (CUDA when available, batched over the
SNR grid); the ToDMA detector runs in numpy on the CPU. Run under
WSL for GPU acceleration. Outputs: data/bertvit_merged.csv.
NFADE fading realisations per pair; ToDMA uses the first 40.
Seed fixed.
"""
from __future__ import annotations
@@ -35,38 +35,20 @@ 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)
DEV = "cuda" if torch.cuda.is_available() else "cpu"
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))
SNRS = np.arange(0.0, 31.0, 2.5)
NFADE = 100 # fading realisations per pair
NFADE_TOD = 40 # ToDMA heavier: first 40 draws
def unit(v):
@@ -87,100 +69,35 @@ def load_pairs():
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 haar_t(n, gen):
G = torch.randn(n, D, D, generator=gen, device=DEV)
Q, R = torch.linalg.qr(G)
return Q * torch.sign(torch.diagonal(R, dim1=-2, dim2=-1)).unsqueeze(-2)
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 aware_batch(t, Q, beta, c, nvar):
"""Batched affinity-aware Wiener demux. t: (b,D) cfloat, Q: (D,D),
c: complex scalar, nvar: (b,) real."""
b = t.shape[0]
g = 1.0 - beta * beta
rho = g * abs(c)**2 / D + nvar # (b,)
Qc = Q.to(torch.cfloat)
A = torch.eye(D, device=DEV, dtype=torch.cfloat) + beta * c * Qc
S = (A @ A.mH / D).unsqueeze(0) \
+ rho.view(b, 1, 1) * torch.eye(D, device=DEV,
dtype=torch.cfloat)
x = torch.linalg.solve(S, t.unsqueeze(-1))
return (A.mH.unsqueeze(0) @ x).squeeze(-1) / D
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
def abscos(a, b):
"""a: (b,D) cfloat, b: (D,) float -> (b,) abs cosine."""
num = (a * b.to(torch.cfloat).conj()).sum(1).abs()
return (num / (a.norm(dim=1) * b.norm())).cpu().numpy()
# ------------------------------------------------------------------
# ToDMA-adapted on real embeddings (complex channel)
# ToDMA-adapted on real embeddings (complex channel, numpy)
# ------------------------------------------------------------------
def todma_prepare(V=1024, T=16):
L = D // T
@@ -238,58 +155,73 @@ def todma_run(tod, codes, h, sig, noise_slots):
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")
print(f"[todma] sparse codes prepared; device = {DEV}")
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}
gen = torch.Generator(device=DEV).manual_seed(SEED)
nb = len(SNRS)
sigs_t = torch.tensor(10 ** (-SNRS / 20.0), device=DEV,
dtype=torch.float32)
keys = ("edma", "oma", "genie", "todma")
res = {k: np.zeros(nb) for k in keys}
cnt = {k: np.zeros(nb) 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]
bi = float(betas[i])
e1 = torch.tensor(A[i], dtype=torch.float32, device=DEV)
e2 = torch.tensor(B[i], dtype=torch.float32, device=DEV)
c1c, c2c = 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)) \
M = haar_t(2, gen)
M1, M2 = M[0], M[1]
Q = M1.T @ M2
h = (torch.randn(2, generator=gen, device=DEV)
+ 1j * torch.randn(2, generator=gen, device=DEV)) \
/ math.sqrt(2)
h1, h2 = h
r0 = h1 * (M1 @ e1) + h2 * (M2 @ e2)
n = (rng.standard_normal(D) + 1j * rng.standard_normal(D)) \
n = (torch.randn(D, generator=gen, device=DEV)
+ 1j * torch.randn(D, generator=gen, device=DEV)) \
/ math.sqrt(2)
n2 = (rng.standard_normal(D) + 1j * rng.standard_normal(D)) \
n2 = (torch.randn(D, generator=gen, device=DEV)
+ 1j * torch.randn(D, generator=gen, device=DEV)) \
/ 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])
r0 = h[0] * (M1 @ e1).to(torch.cfloat) \
+ h[1] * (M2 @ e2).to(torch.cfloat)
r = r0.unsqueeze(0) + sigs_t.view(-1, 1) * n.unsqueeze(0)
t1 = (M1.T.to(torch.cfloat) @ r.unsqueeze(-1)).squeeze(-1) / h[0]
t2 = (M2.T.to(torch.cfloat) @ r.unsqueeze(-1)).squeeze(-1) / h[1]
c1 = (h[1] / h[0]).item()
c2 = (h[0] / h[1]).item()
v1 = sigs_t**2 / h[0].abs()**2
v2 = sigs_t**2 / h[1].abs()**2
g1 = aware_batch(t1, Q, bi, c1, v1)
g2 = aware_batch(t2, Q.T, bi, c2, v2)
res["edma"] += 0.5 * (abscos(g1, e1) + abscos(g2, e2))
o1 = e1.to(torch.cfloat).unsqueeze(0) \
+ math.sqrt(2) * sigs_t.view(-1, 1) * n.unsqueeze(0) / h[0]
o2 = e2.to(torch.cfloat).unsqueeze(0) \
+ math.sqrt(2) * sigs_t.view(-1, 1) * n2.unsqueeze(0) / h[1]
res["oma"] += 0.5 * (abscos(o1, e1) + abscos(o2, e2))
ge1 = (M1.T.to(torch.cfloat)
@ (r - h[1] * (M2 @ e2).to(torch.cfloat)).unsqueeze(-1)
).squeeze(-1) / h[0]
ge2 = (M2.T.to(torch.cfloat)
@ (r - h[0] * (M1 @ e1).to(torch.cfloat)).unsqueeze(-1)
).squeeze(-1) / h[1]
res["genie"] += 0.5 * (abscos(ge1, e1) + abscos(ge2, e2))
for kk in ("edma", "oma", "genie"):
cnt[kk] += 1
if f < NFADE_TOD:
hnp = (complex(h[0].item()), complex(h[1].item()))
nslots = [(rng.standard_normal(tod[4])
+ 1j * rng.standard_normal(tod[4]))
/ math.sqrt(2) for _ in range(tod[3])]
e1n, e2n = A[i], B[i]
for k, s in enumerate(SNRS):
sig = 10 ** (-s / 20.0)
recs = todma_run(tod, (c1c, c2c), hnp, sig, nslots)
got = [cosine(recs[j], (e1n, e2n)[j])
for j in range(2) if recs[j] is not None]
if got:
res["todma"][k] += float(np.mean(got))
@@ -299,23 +231,6 @@ def main():
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))
@@ -323,8 +238,7 @@ def main():
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}) "
print(f" {s:4.1f} dB EDMA {res['edma'][k]:.3f} "
f"ToDMA {res['todma'][k]:.3f} OMA {res['oma'][k]:.3f} "
f"genie {res['genie'][k]:.3f}")
+140 -119
View File
@@ -1,23 +1,23 @@
"""
Capacity-matched EDMA refinement (parameter budget equal to the
attention scheme: 4 d^2 = 2.36M at d = 768).
Refinement stage for the v2 (affinity-aware Wiener) EDMA receiver.
================================================================
Four-head averaged gated refinement applied to the closed-form
demultiplexer output:
Trains the single-gate refinement operator
out = (1/4) sum_k D softmax(Q_k x / sqrt(D)) .* x,
out = D softmax(W z / sqrt(D)) .* z, z = Re(e_hat),
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.
(0.59M parameters at d = 768) on aware-demultiplexer outputs, then
warm-starts a capacity-check variant with four heads (4 d^2 = 2.36M)
and fine-tunes it, so the family contains the single gate by
construction. Training data: parametric pairs at beta = 0.028, a
fixed pool of 32 independent Haar mask pairs, Rayleigh channels,
complex noise, training SNR uniform in [5, 25] dB, Adam 5e-4 with
gradient clipping, batch 48, 220 epochs (stage 2 from epoch 120 at
lr 2e-4).
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.
NFADE fading draws per pair. Appends columns `edma_ref` (single
gate) and `edma_ref2` (four heads) to data/bertvit_merged.csv.
Requires torch (run under WSL with CUDA if available).
"""
from __future__ import annotations
import csv
@@ -26,161 +26,182 @@ import time
import numpy as np
import torch
from fig_real_merged import load_pairs, cosine, SNRS, NFADE, D, DATA
from fig_real_merged import load_pairs, SNRS, NFADE, D, DATA
SEED = 2026
rng = np.random.default_rng(SEED + 31)
torch.manual_seed(SEED + 31)
DEV = "cuda" if torch.cuda.is_available() else "cpu"
BETA0 = 0.028
G0 = 1.0 - BETA0**2
print(f"[refine] device = {DEV}")
def haar_t(gen):
Q, R = torch.linalg.qr(torch.randn(D, D, generator=gen))
return Q * torch.sign(torch.diagonal(R))
def haar_t(n, gen):
G = torch.randn(n, D, D, generator=gen, device=DEV)
Q, R = torch.linalg.qr(G)
return Q * torch.sign(torch.diagonal(R, dim1=-2, dim2=-1)).unsqueeze(-2)
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]
def aware_t(t1, Q, beta, c1, nvar):
"""Batched affinity-aware Wiener demux in torch (complex)."""
b = t1.shape[0]
g = 1.0 - beta * beta
rho = g * (c1.abs()**2) / D + nvar # (b,)
A = torch.eye(D, device=DEV, dtype=torch.cfloat).expand(b, D, D) \
+ beta * c1.view(b, 1, 1) * Q.to(torch.cfloat)
S = A @ A.mH / D + rho.view(b, 1, 1) \
* torch.eye(D, device=DEV, dtype=torch.cfloat)
x = torch.linalg.solve(S, t1.unsqueeze(-1))
return (A.mH @ x).squeeze(-1) / D
def train_batch(masks, Qs, gen, batch):
"""Generate one training batch of aware-demux outputs (user 1)."""
e1 = torch.nn.functional.normalize(
torch.randn(batch, D, generator=gen, device=DEV), dim=1)
w = torch.randn(batch, D, generator=gen, device=DEV)
w = w - (w * e1).sum(1, keepdim=True) * e1
w = torch.nn.functional.normalize(w, dim=1)
e2 = BETA0 * e1 + math.sqrt(G0) * w
sel = torch.randint(len(masks), (batch,), generator=gen, device=DEV)
M1 = masks[0][sel]; M2 = masks[1][sel]; Q = Qs[sel]
snr = 5.0 + 20.0 * torch.rand(batch, generator=gen, device=DEV)
sig = 10 ** (-snr / 20.0)
h = (torch.randn(batch, 2, generator=gen, device=DEV)
+ 1j * torch.randn(batch, 2, generator=gen, device=DEV)) \
/ math.sqrt(2)
n = (torch.randn(batch, D, generator=gen, device=DEV)
+ 1j * torch.randn(batch, D, generator=gen, device=DEV)) \
/ math.sqrt(2)
r = h[:, :1] * (M1 @ e1.unsqueeze(-1)).squeeze(-1).to(torch.cfloat) \
+ h[:, 1:2] * (M2 @ e2.unsqueeze(-1)).squeeze(-1).to(torch.cfloat) \
+ sig.view(-1, 1) * n
t1 = (M1.transpose(-1, -2).to(torch.cfloat)
@ r.unsqueeze(-1)).squeeze(-1) / h[:, :1]
c1 = h[:, 1] / h[:, 0]
nvar = sig**2 / h[:, 0].abs()**2
g1 = aware_t(t1, Q, BETA0, c1, nvar)
return g1.real.float(), e1
def train_refiners(epochs=220, steps=20, batch=48, lr=5e-4,
l2=0.5, l3=0.5, pool=32, stage2_at=120):
print(f"=== training refinement (single gate {D*D/1e6:.2f}M, "
f"then 4-head warm start {4*D*D/1e6:.2f}M) ===", flush=True)
gen = torch.Generator(device=DEV).manual_seed(SEED + 31)
U1 = haar_t(pool, gen); U2 = haar_t(pool, gen)
masks = (U1, U2)
Qs = U1.transpose(-1, -2) @ U2
params = [torch.nn.Parameter(
torch.randn(D, D, generator=gen, device=DEV) / math.sqrt(D))]
opt = torch.optim.Adam(params, lr=lr)
stage2_at = 120 # epochs of single-gate pre-training
P_single = None
def forward(x):
def forward(x, ps):
outs = [D * torch.softmax((x @ Qk.T) / math.sqrt(D), dim=1) * x
for Qk in params]
return sum(outs) / len(params)
for Qk in ps]
return sum(outs) / len(ps)
t0 = time.time()
for ep in range(epochs):
if ep == stage2_at:
P_single = params[0].detach().clone()
base = params[0].detach()
params = [torch.nn.Parameter(
base.clone() + 0.02 * torch.randn(D, D, generator=gen)
base.clone() + 0.02 * torch.randn(D, D, generator=gen,
device=DEV)
/ 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)
print(f" [warm start] 4 heads 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()
with torch.no_grad():
x, tgt = train_batch(masks, Qs, gen, batch)
out = forward(x, params)
mse = ((out - tgt)**2).mean()
cs = torch.nn.functional.cosine_similarity(out, tgt, 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:
if (ep + 1) % 40 == 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]
return P_single, [p.detach() 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 refine_apply(ps, z):
"""z: (b, D) real torch tensor; ps: list of gates."""
outs = [D * torch.softmax((z @ Qk.T) / math.sqrt(D), dim=1) * z
for Qk in ps]
return sum(outs) / len(ps)
def main():
A, B, betas = load_pairs()
P = train_refiner2()
ref = np.zeros(len(SNRS)); cnt = 0
P1, P4 = train_refiners()
gen = torch.Generator(device=DEV).manual_seed(SEED + 77)
ref1 = np.zeros(len(SNRS)); ref4 = np.zeros(len(SNRS)); cnt = 0
t0 = time.time()
At = torch.tensor(A, dtype=torch.float32, device=DEV)
Bt = torch.tensor(B, dtype=torch.float32, device=DEV)
for i in range(len(A)):
e1, e2, bi = A[i], B[i], float(betas[i])
gi = 1.0 - bi**2
bi = float(betas[i])
e1 = At[i]; e2 = Bt[i]
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)) \
M = haar_t(2, gen)
M1, M2 = M[0], M[1]
Q = M1.T @ M2
h = (torch.randn(2, generator=gen, device=DEV)
+ 1j * torch.randn(2, generator=gen, device=DEV)) \
/ math.sqrt(2)
h1, h2 = h
r0 = h1 * (M1 @ e1) + h2 * (M2 @ e2)
n = (rng.standard_normal(D) + 1j * rng.standard_normal(D)) \
n = (torch.randn(D, generator=gen, device=DEV)
+ 1j * torch.randn(D, generator=gen, device=DEV)) \
/ 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))
r0 = h[0] * (M1 @ e1).to(torch.cfloat) \
+ h[1] * (M2 @ e2).to(torch.cfloat)
sigs = torch.tensor(10 ** (-SNRS / 20.0), device=DEV,
dtype=torch.float32)
nb = len(SNRS)
r = r0.unsqueeze(0) + sigs.view(-1, 1) * n.unsqueeze(0)
t1 = (M1.T.to(torch.cfloat) @ r.unsqueeze(-1)).squeeze(-1) / h[0]
t2 = (M2.T.to(torch.cfloat) @ r.unsqueeze(-1)).squeeze(-1) / h[1]
c1 = (h[1] / h[0]).expand(nb)
c2 = (h[0] / h[1]).expand(nb)
v1 = (sigs**2 / h[0].abs()**2)
v2 = (sigs**2 / h[1].abs()**2)
g1 = aware_t(t1, Q.expand(nb, D, D), bi, c1, v1).real.float()
g2 = aware_t(t2, Q.T.expand(nb, D, D), bi, c2, v2).real.float()
with torch.no_grad():
for P, acc in ((([P1]), ref1), ((P4), ref4)):
o1 = refine_apply(P, g1)
o2 = refine_apply(P, g2)
cs1 = torch.nn.functional.cosine_similarity(
o1, e1.unsqueeze(0), dim=1).abs()
cs2 = torch.nn.functional.cosine_similarity(
o2, e2.unsqueeze(0), dim=1).abs()
acc += (0.5 * (cs1 + cs2)).cpu().numpy()
cnt += 1
print(f" pair {i+1}/{len(A)} done ({time.time()-t0:.0f}s)",
flush=True)
ref /= cnt
ref1 /= cnt; ref4 /= 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 col in ("edma_ref", "edma_ref2"):
if col not in names:
names.append(col)
for k, r in enumerate(rows):
r["edma_ref2"] = f"{ref[k]}"
r["edma_ref"] = f"{ref1[k]}"
r["edma_ref2"] = f"{ref4[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")
print("[OK] appended edma_ref / 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}")
print(f" {float(r['snr_db']):4.1f} dB "
f"EDMA {float(r['edma']):.3f} ref {ref1[k]:.3f} "
f"ref2 {ref4[k]:.3f} genie {float(r['genie']):.3f}")
if __name__ == "__main__":
+179
View File
@@ -0,0 +1,179 @@
"""Canonical figure rendering. Reads ONLY data/*.csv, writes fig/*.pdf.
Figures: fig_floor, fig_rate_corrected, fig_beta_sweep_corrected,
fig_sic, fig_multiuser_corrected. (fig_bertvit_merged is rendered by
replot_merged.py; block_diagram.pdf comes from block_diagram_src.tex.)
One physical geometry and one label dictionary for every plot.
"""
import csv
import math
from pathlib import Path
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)
LBL = {
"edma": "EDMA",
"blind": "Affinity-blind",
"oma": "OMA",
"genie": "Genie-aided SIC bound",
"sic": "Realizable analog SIC",
"todma": "ToDMA-adapted",
"mac": "MAC sum capacity",
"coop": "Full-cooperation bound",
"hybrid": "EDMA + refinement stage",
}
def rows_of(name):
return list(csv.DictReader(open(DATA / f"{name}.csv")))
def col(rows, k):
return [float(r[k]) for r in rows]
def save(fig, name):
fig.subplots_adjust(**AXES_RECT)
fig.savefig(FIG / f"{name}.pdf")
plt.close(fig)
print(f"[OK] wrote {name}.pdf")
# ------------------------------------------------------ fig_floor
def fig_floor():
rows = rows_of("floor_validation")
fig, ax = plt.subplots()
colors = {"256": "C0", "768": "C3"}
beta = 0.311
for d in ("256", "768"):
rd = [r for r in rows if r["d"] == d or r["d"] == f"{d}.0"
or float(r["d"]) == float(d)]
snr = col(rd, "snr_db")
ax.plot(snr, col(rd, "mse_mc"), "o", ms=3.5, color=colors[d],
mfc="none", label=rf"Monte Carlo, $d={d}$")
ax.plot(snr, col(rd, "mse_theory"), "-", color=colors[d],
label=rf"Theorem 1, $d={d}$")
if d == "768":
ax.plot(snr, col(rd, "mse_blind"), "--", color="C1", lw=1.2,
label=LBL["blind"])
g = 1.0 - beta**2
ax.axhline(math.sqrt(g) / 2, color="gray", lw=0.8, ls="--")
ax.axhline(0.5, color="gray", lw=0.8, ls=":")
ax.annotate("blind floor $1/2$", xy=(17.0, 0.512), fontsize=7,
color="gray")
ax.annotate(r"aware floor $\sqrt{1-\beta^2}/2$", xy=(14.0, 0.432),
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.4, 1.05)
ax.legend(loc="lower left", bbox_to_anchor=(0.02, 0.18))
save(fig, "fig_floor")
# ------------------------------------------------ fig_rate_corrected
def fig_rate():
rows = rows_of("rate_corrected")
snr = col(rows, "snr_db")
fig, ax = plt.subplots()
ax.plot(snr, col(rows, "edma"), "-", color="C3", label=LBL["edma"])
ax.plot(snr, col(rows, "blind"), ":", color="C4", lw=1.2,
label=LBL["blind"])
ax.plot(snr, col(rows, "oma"), "--", color="C1", label=LBL["oma"])
ax.plot(snr, col(rows, "genie"), "-.", color="C0", label=LBL["genie"])
ax.plot(snr, col(rows, "mac"), "-", color="k", lw=1.0, label=LBL["mac"])
ax.set_xlabel("Per-block SNR $\\rho$ [dB]")
ax.set_ylabel("Effective sum rate [bps/Hz]")
ax.set_xlim(0, 40); ax.set_ylim(0, 3.2)
ax.legend(loc="upper left")
save(fig, "fig_rate_corrected")
# ------------------------------------------ fig_beta_sweep_corrected
def fig_beta_sweep():
rows = rows_of("beta_sweep_corrected")
fig, ax = plt.subplots()
for s, cc in (("10", "C0"), ("20", "C3")):
rd = [r for r in rows if float(r["snr_db"]) == float(s)]
b = col(rd, "beta")
ax.plot(b, col(rd, "edma"), "-", color=cc,
label=rf"EDMA, $\rho={s}$ dB")
ax.axhline(float(rd[0]["blind"]), color=cc, ls=":", lw=1.0)
ax.axhline(float(rd[0]["oma"]), color=cc, ls="--", lw=1.0)
ax.axhline(float(rd[0]["genie"]), color=cc, ls="-.", lw=0.8)
# one legend entry per reference style (color-independent)
ax.plot([], [], ls=":", color="gray", label=LBL["blind"])
ax.plot([], [], ls="--", color="gray", label=LBL["oma"])
ax.plot([], [], ls="-.", color="gray", label=LBL["genie"])
for b0 in (0.030, 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.0)
ax.legend(loc="upper left")
save(fig, "fig_beta_sweep_corrected")
# ------------------------------------------------------- fig_sic
def fig_sic():
rows = rows_of("sic_comparison")
snr = col(rows, "snr_db")
fig, ax = plt.subplots()
ax.plot(snr, col(rows, "edma"), "o-", color="C3", label=LBL["edma"])
ax.plot(snr, col(rows, "blind"), "d:", color="C4", label=LBL["blind"])
ax.plot(snr, col(rows, "sic"), "^-.", color="C2", label=LBL["sic"])
ax.plot(snr, col(rows, "oma"), "v--", color="C1", label=LBL["oma"])
ax.plot(snr, col(rows, "genie"), "-", color="gray", lw=1.0,
label=LBL["genie"])
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")
save(fig, "fig_sic")
# ------------------------------------------ fig_multiuser_corrected
def fig_multiuser():
rows = rows_of("multiuser_corrected")
fig, ax = plt.subplots()
colors = {"2": "C0", "3": "C2", "4": "C3"}
for U in ("2", "3", "4"):
rd = [r for r in rows if float(r["U"]) == float(U)]
snr = col(rd, "snr_db")
ax.plot(snr, col(rd, "edma_mc"), "-", color=colors[U],
label=rf"EDMA, $U={U}$")
ax.plot(snr, col(rd, "oma"), "--", color=colors[U], lw=1.0,
label=rf"OMA, $U={U}$")
mk = [i for i, s in enumerate(snr) if s % 5 == 0]
ax.plot([snr[i] for i in mk], [col(rd, "edma_mc")[i] for i in mk],
"o", color=colors[U], ms=4, mfc="none")
ax.set_xlabel("Per-block SNR $\\rho$ [dB]")
ax.set_ylabel("Effective sum rate [bps/Hz]")
ax.set_xlim(0, 30)
ax.legend(loc="upper left")
save(fig, "fig_multiuser_corrected")
if __name__ == "__main__":
import sys
todo = set(sys.argv[1:])
ALL = {"floor": fig_floor, "rate": fig_rate, "beta": fig_beta_sweep,
"sic": fig_sic, "multi": fig_multiuser}
for name, fn in ALL.items():
if not todo or name in todo:
fn()
+3 -2
View File
@@ -1,6 +1,7 @@
"""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."""
The capacity-check column edma_ref2 remains in the CSV but is not
plotted (it tracks edma_ref; quoted in the text only)."""
import csv
from pathlib import Path
import matplotlib
@@ -24,7 +25,7 @@ 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"), "o-", color="C3", label="EDMA")
ax.plot(snr, col("edma_ref"), "^-", color="C2",
label="EDMA + refinement stage")
ax.plot(snr, col("todma"), "d-.", color="C4", label="ToDMA-adapted")
-43
View File
@@ -1,43 +0,0 @@
"""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}")
+182 -449
View File
@@ -1,26 +1,45 @@
"""
Revision simulations for the EDMA TCOM resubmission.
Simulations for the EDMA TVT manuscript (v2 design).
=======================================================
Implements the per-realisation (finite-d) analysis and the corrected
energy-normalised rate accounting, plus the reviewer-requested
experiments:
Design v2: each user applies an independent orthogonal mask; the
receiver runs one matched filter per user followed by the
affinity-aware linear MMSE demultiplexer, which exploits the
coherent interference component that the pairwise affinity beta
predicts. Per-realization statistic for user 1 (c1 = h2/h1):
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
t1 = (I + beta*c1*Q) e1 + sqrt(g)*c1*Q w + n_t, Q = M1^T M2,
Conventions (identical to the revised manuscript):
and the demultiplexer is the Wiener filter
e1_hat = (1/d) A^H (A A^H/d + (g|c1|^2/d + sig^2/|h1|^2) I)^{-1} t1,
A = I + beta*c1*Q, g = 1 - beta^2.
Closed form (Theorem 1, d -> inf, per channel realization):
MSE_1 = rho_e / sqrt((1 + beta^2|c1|^2 + rho_e)^2 - 4 beta^2|c1|^2),
rho_e = g|c1|^2 + d sig^2/|h1|^2; floor at |c1| = 1: sqrt(g)/2.
The affinity-blind receiver (beta = 0 in the filter) reduces to a
scalar shrinkage of the matched filter with floor 1/2, so the entire
cosine gain of the aware receiver is attributable to the predicted
affinity. Effective SINR: eta = 1/MSE - 1 (biased MMSE convention).
Experiments in this file (CPU, numpy):
E0 theorem_check : closed form vs Monte Carlo, both users
E1 fig_floor : per-user MSE vs block SNR, aware vs blind floor
E7a rate_corrected + beta_sweep_corrected : closed-form rate curves
The Monte Carlo experiments E2, E3, E4, E5, E7c, E8, E9 are canonical
in revision_sims_gpu.py (torch backend, run under WSL); figures are
rendered from data/ by replot_all.py and replot_merged.py.
Conventions (identical to the 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.
Fixed seed 2026. CSVs -> ../data, PDFs -> ../fig.
"""
from __future__ import annotations
import csv
@@ -39,13 +58,27 @@ 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,
"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)
# shared legend-label dictionary (single source for every figure)
LBL = {
"edma": "EDMA",
"blind": "Affinity-blind",
"oma": "OMA",
"genie": "Genie-aided SIC bound",
"sic": "Realizable analog SIC",
"todma": "ToDMA-adapted",
"mac": "MAC sum capacity",
"hybrid": "EDMA + refinement stage",
"haar": "Haar masks",
"wh": "Walsh-Hadamard masks",
}
rng = np.random.default_rng(2026)
@@ -86,69 +119,80 @@ def embed_pair(d, beta):
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 cnoise(d):
return (rng.standard_normal(d) + 1j * rng.standard_normal(d)) / math.sqrt(2)
def cosine(a, b):
return abs(np.vdot(a, b)) / (np.linalg.norm(a) * np.linalg.norm(b))
def mse_theory(beta, c1, rho_e):
"""Theorem 1: per-realization MSE of the aware demultiplexer."""
a0 = 1.0 + beta**2 * abs(c1)**2 + rho_e
return rho_e / math.sqrt(a0 * a0 - 4.0 * beta**2 * abs(c1)**2)
def mse_blind(c1, dsig2_h):
"""Affinity-blind scalar-shrinkage MSE (beta = 0 in the filter)."""
r0 = abs(c1)**2 + dsig2_h
return r0 / (1.0 + r0)
def eta_of(mse):
"""Effective SINR of a (possibly biased) estimator with unit signal."""
return 1.0 / mse - 1.0
def aware(t1, Q, beta, c1, nvar, d):
"""Affinity-aware Wiener demultiplexer applied to t1 = M1^T r / h1.
Uses A A^H = (1+beta^2|c1|^2) I + beta(c1 Q + conj(c1) Q^T), so the
system matrix is assembled in O(d^2) and solved with one LU."""
g = 1.0 - beta * beta
rho = g * abs(c1)**2 / d + nvar
S = beta * (c1 * Q + np.conj(c1) * Q.T) / d
S[np.diag_indices(d)] += (1.0 + beta**2 * abs(c1)**2) / d + rho
x = np.linalg.solve(S, t1)
return (x + beta * np.conj(c1) * (Q.T @ x)) / d
def blind(t1, c1, nvar, d):
"""Affinity-blind receiver: scalar shrinkage of the matched filter."""
lam = (1.0 / d) / (1.0 / d + abs(c1)**2 / d + nvar)
return lam * t1
# ------------------------------------------------------------------
# E0 : Theorem-1 verification
# E0 : Theorem-1 verification (both users, random phases)
# ------------------------------------------------------------------
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 ===")
def E0_theorem_check(d=512, betas=(0.0, 0.311, 0.5, 0.7), ntr=200, snr=20.0):
print("\n=== E0: Theorem 1 (aware-demultiplexer MSE) verification ===")
sig = 10 ** (-snr / 20.0)
rows = []
worst = 0.0
for beta in betas:
# random unit-modulus channels (AWGN-type magnitude, random phase)
errs1, errs2 = [], []
g = 1.0 - beta**2
r1s, r2s = [], []
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))
M1, M2 = haar(d), haar(d)
Q = M1.T @ M2
r = h1 * (M1 @ e1) + h2 * (M2 @ e2) + sig * cnoise(d)
c1, c2 = h2 / h1, h1 / h2
g1 = aware(M1.T @ r / h1, Q, beta, c1, sig**2 / abs(h1)**2, d)
g2 = aware(M2.T @ r / h2, Q.T, beta, c2, sig**2 / abs(h2)**2, d)
th1 = mse_theory(beta, c1, g * abs(c1)**2 + d * sig**2 / abs(h1)**2)
th2 = mse_theory(beta, c2, g * abs(c2)**2 + d * sig**2 / abs(h2)**2)
r1s.append(np.linalg.norm(g1 - e1)**2 / th1)
r2s.append(np.linalg.norm(g2 - e2)**2 / th2)
r1, r2 = float(np.mean(r1s)), float(np.mean(r2s))
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}"
@@ -161,12 +205,11 @@ def E0_theorem_check(d=512, betas=(0.0, 0.311, 0.5, 0.7), ntr=300):
# ------------------------------------------------------------------
# E1 : interference floor (MSE vs block SNR), AWGN point |h|=1
# E1 : MSE vs block SNR at |h|=1 -- aware floor sqrt(g)/2 vs blind 1/2
# ------------------------------------------------------------------
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 ===")
def E1_floor(beta=0.311, dims=(256, 768), snr_db=np.arange(0, 41, 2.5), ntr=120):
print("\n=== E1: finite-d validation, aware vs blind floor ===")
g = 1.0 - beta**2
csi = C_SI(beta, 1.0 + 0j)
fig, ax = plt.subplots()
colors = {256: "C0", 768: "C3"}
rows = []
@@ -174,444 +217,134 @@ def E1_floor(beta=0.311, dims=(256, 768), snr_db=np.arange(0, 41, 2.5), ntr=150)
mc = np.zeros(len(snr_db))
for _ in range(ntr):
e1, e2 = embed_pair(d, beta)
M1, M2 = two_user_masks(d, beta)
M1, M2 = haar(d), haar(d)
Q = M1.T @ M2
r0 = (M1 @ e1) + (M2 @ e2)
n = (rng.standard_normal(d) + 1j * rng.standard_normal(d)) / math.sqrt(2)
n = cnoise(d)
for k, s in enumerate(snr_db):
sig = 10 ** (-s / 20.0)
g1, _ = demux(r0 + sig * n, M1, M2, 1.0, 1.0, beta)
g1 = aware(M1.T @ (r0 + sig * n), Q, beta, 1.0, sig**2, d)
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)
th = np.array([mse_theory(beta, 1.0, g + d / r) for r in rho])
bl = np.array([mse_blind(1.0, d / r) for r in rho])
ax.semilogy(snr_db, mc, "o", ms=3.5, color=colors[d], mfc="none",
label=rf"MC, $d={d}$")
label=rf"Monte Carlo, $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.semilogy(snr_db, bl, "--", color="C1", lw=1.1,
label=LBL["blind"])
for s, m, t, b in zip(snr_db, mc, th, bl):
rows.append([d, s, m, t, b])
dev = 100 * max(abs(mc / th - 1))
print(f" d={d}: max MC/theory dev {dev:.1f}%")
ax.axhline(math.sqrt(g) / 2, color="gray", lw=0.8, ls="--")
ax.axhline(0.5, color="gray", lw=0.8, ls=":")
ax.text(1.0, 0.52, r"blind floor $1/2$", fontsize=7, color="gray")
ax.text(22.0, 0.40, r"aware floor $\sqrt{1-\beta^2}/2$",
fontsize=7, color="gray")
ax.set_yscale("linear")
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.set_xlim(0, 40); ax.set_ylim(0.4, 1.05)
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)
write_csv("floor_validation",
["d", "snr_db", "mse_mc", "mse_theory", "mse_blind"], rows)
print(f" aware floor {math.sqrt(g)/2:.4f} vs blind floor 0.5000 "
f"(ratio {0.5/(math.sqrt(g)/2):.4f} = 1/sqrt(1-beta^2))")
# ------------------------------------------------------------------
# E2 : realisable SIC vs genie SIC vs EDMA vs OMA (Rayleigh)
# E7 : effective-rate figures (eta = 1/MSE - 1)
# ------------------------------------------------------------------
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}")
def T_edma(rho, d, beta):
m = mse_theory(beta, 1.0, (1.0 - beta**2) + d / rho)
return 2.0 * math.log2(1.0 + eta_of(m))
# ------------------------------------------------------------------
# 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 T_blind(rho, d):
return 2.0 * math.log2(1.0 + 1.0 / (1.0 + d / rho))
def E7_rates(beta=0.311, d=512):
print("\n=== E7a: corrected effective-rate comparison ===")
snr_db = np.arange(0, 31, 1.0)
print("\n=== E7a: effective-rate comparison ===")
snr_db = np.arange(0, 41, 0.5)
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)
Te = np.array([T_edma(r, d, beta) for r in rho])
Tb = np.array([T_blind(r, d) for r in rho])
To = 2 * np.log2(1 + rho / (2 * d))
Tg = 2 * np.log2(1 + rho / d)
Cm = 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.plot(snr_db, Te, "-", color="C3", label=LBL["edma"])
ax.plot(snr_db, Tb, ":", color="C4", lw=1.2, label=LBL["blind"])
ax.plot(snr_db, To, "--", color="C1", label=LBL["oma"])
ax.plot(snr_db, Tg, "-.", color="C0", label=LBL["genie"])
ax.plot(snr_db, Cm, "-", color="k", lw=1.0, label=LBL["mac"])
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.set_xlim(0, 40); 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]]
rows = [[s, Te[i], Tb[i], To[i], Tg[i], Cm[i]]
for i, s in enumerate(snr_db)]
write_csv("rate_corrected",
["snr_db", "edma", "edma_ideal", "oma", "genie", "mac"], rows)
["snr_db", "edma", "blind", "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(f" at 20 dB: EDMA {Te[i20]:.3f}, blind {Tb[i20]:.3f}, "
f"OMA {To[i20]:.3f} (gain {Te[i20]/To[i20]:.2f}x), "
f"MAC {Cm[i20]:.3f}, EDMA/MAC {Te[i20]/Cm[i20]:.3f}")
g = 1.0 - beta**2
rho_c = 2 * d * (2 / math.sqrt(g) - 1)
ix = np.where(To >= Te)[0]
rc_num = snr_db[ix[0]] if len(ix) else float("nan")
print(f" OMA re-crossover: floor formula {10*math.log10(rho_c):.1f} dB, "
f"numerical {rc_num:.1f} dB "
f"(blind: {10*math.log10(2*d):.1f} dB)")
print("\n=== E7b: corrected beta sweep ===")
print("\n=== E7b: value-of-affinity 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)
Te = np.array([T_edma(rho_s, d, b) for b in betas])
Tb = T_blind(rho_s, d)
To = 2 * math.log2(1 + rho_s / (2 * d))
Tg = 2 * math.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)")
ax.axhline(Tb, color=col, ls=":", lw=1.0)
ax.axhline(To, color=col, ls="--", lw=1.0)
ax.axhline(Tg, color=col, ls="-.", lw=0.8)
ixg = np.where(Te >= Tg)[0]
bg = betas[ixg[0]] if len(ixg) else float("nan")
print(f" rho={s} dB: EDMA(0)/blind = {Te[0]/Tb:.3f}, "
f"EDMA(0.311) gain over blind "
f"{Te[np.argmin(abs(betas-0.311))]/Tb:.3f}x, "
f"crosses genie at beta ~ {bg:.2f}")
for i, b in enumerate(betas):
rows.append([s, b, Te[i], To, Tg])
for b0 in (0.031, 0.311):
rows.append([s, b, Te[i], Tb, To, Tg])
for b0 in (0.030, 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)
ax.set_xlim(0, 1)
ax.legend(loc="upper left")
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)
["snr_db", "beta", "edma", "blind", "oma", "genie"], 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,
"E0": E0_theorem_check, "E1": E1_floor, "E7a": E7_rates,
}
for name, fn in ALL.items():
if not todo or name in todo:
fn()
print("\nAll requested revision simulations complete.")
print("\nAll requested simulations complete.")
+416
View File
@@ -0,0 +1,416 @@
"""
GPU-accelerated Monte Carlo experiments (torch backend).
========================================================
Computes the CSV artifacts of experiments E2, E3, E4, E5, E7c, E8
of revision_sims.py with identical models and conventions, using
torch (CUDA when available) for the dense linear algebra. All
random draws come from the numpy generator with the documented seed
2026, so the sample stream is platform-independent; torch only
accelerates QR, matrix products, and linear solves in float32 /
complex64 precision. Figures are rendered separately by
replot_all.py, which reads only data/.
Run under WSL: python3 revision_sims_gpu.py E2 E3 E4 E5 E7c E8
"""
from __future__ import annotations
import csv
import math
import sys
import time
from pathlib import Path
import numpy as np
import torch
ROOT = Path(__file__).resolve().parents[1]
CSV_DIR = ROOT / "data"
SEED = 2026
rng = np.random.default_rng(SEED)
DEV = "cuda" if torch.cuda.is_available() else "cpu"
print(f"[gpu] device = {DEV}")
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}")
def haar_g(d):
"""Haar orthogonal on the GPU from a numpy Gaussian draw."""
G = torch.tensor(rng.standard_normal((d, d)), dtype=torch.float32,
device=DEV)
Q, R = torch.linalg.qr(G)
return Q * torch.sign(torch.diagonal(R)).unsqueeze(0)
def embed_pair(d, beta):
e1 = rng.standard_normal(d)
e1 /= np.linalg.norm(e1)
w = rng.standard_normal(d)
w -= (w @ e1) * e1
w /= np.linalg.norm(w)
e2 = beta * e1 + math.sqrt(1.0 - beta**2) * w
return (torch.tensor(e1, dtype=torch.float32, device=DEV),
torch.tensor(e2, dtype=torch.float32, device=DEV))
def rayleigh2():
h = (rng.standard_normal(2) + 1j * rng.standard_normal(2)) / math.sqrt(2)
return complex(h[0]), complex(h[1])
def cnoise_g(d):
n = (rng.standard_normal(d) + 1j * rng.standard_normal(d)) / math.sqrt(2)
return torch.tensor(n, dtype=torch.complex64, device=DEV)
def aware_g(t, Q, beta, c, nvar):
"""Batched aware Wiener demux. t: (b,d) cfloat, Q: (d,d) float,
c: python complex, nvar: (b,) tensor."""
d = Q.shape[0]
b = t.shape[0]
g = 1.0 - beta * beta
rho = g * abs(c)**2 / d + nvar
A = torch.eye(d, device=DEV, dtype=torch.complex64) \
+ beta * c * Q.to(torch.complex64)
S = (A @ A.mH / d).unsqueeze(0) \
+ rho.view(b, 1, 1) * torch.eye(d, device=DEV,
dtype=torch.complex64)
x = torch.linalg.solve(S, t.unsqueeze(-1))
return (A.mH.unsqueeze(0) @ x).squeeze(-1) / d
def abscos(a, e):
num = (a * e.to(a.dtype).conj()).sum(-1).abs()
return (num / (a.norm(dim=-1) * e.norm())).cpu().numpy()
# ------------------------------------------------------------------
def E2_sic(beta=0.311, d=512, ntr=400):
print("\n=== E2: receiver comparison under block-Rayleigh fading ===")
snr_db = np.arange(0, 31, 2.5)
sigs = torch.tensor(10 ** (-snr_db / 20.0), dtype=torch.float32,
device=DEV)
nb = len(snr_db)
res = {k: np.zeros(nb) for k in ("edma", "blind", "oma", "genie", "sic")}
t0 = time.time()
for tr in range(ntr):
e1, e2 = embed_pair(d, beta)
M1, M2 = haar_g(d), haar_g(d)
Q = M1.T @ M2
h1, h2 = rayleigh2()
n = cnoise_g(d); n2 = cnoise_g(d)
r0 = h1 * (M1 @ e1).to(torch.complex64) \
+ h2 * (M2 @ e2).to(torch.complex64)
r = r0.unsqueeze(0) + sigs.view(-1, 1) * n.unsqueeze(0)
t1 = (M1.T.to(torch.complex64) @ r.unsqueeze(-1)).squeeze(-1) / h1
t2 = (M2.T.to(torch.complex64) @ r.unsqueeze(-1)).squeeze(-1) / h2
c1, c2 = h2 / h1, h1 / h2
v1 = sigs**2 / abs(h1)**2
v2 = sigs**2 / abs(h2)**2
g1 = aware_g(t1, Q, beta, c1, v1)
g2 = aware_g(t2, Q.T, beta, c2, v2)
res["edma"] += 0.5 * (abscos(g1, e1) + abscos(g2, e2))
res["blind"] += 0.5 * (abscos(t1, e1) + abscos(t2, e2))
o1 = e1.to(torch.complex64).unsqueeze(0) \
+ math.sqrt(2) * sigs.view(-1, 1) * n.unsqueeze(0) / h1
o2 = e2.to(torch.complex64).unsqueeze(0) \
+ math.sqrt(2) * sigs.view(-1, 1) * n2.unsqueeze(0) / h2
res["oma"] += 0.5 * (abscos(o1, e1) + abscos(o2, e2))
ge1 = (M1.T.to(torch.complex64)
@ (r - h2 * (M2 @ e2).to(torch.complex64)).unsqueeze(-1)
).squeeze(-1) / h1
ge2 = (M2.T.to(torch.complex64)
@ (r - h1 * (M1 @ e1).to(torch.complex64)).unsqueeze(-1)
).squeeze(-1) / h2
res["genie"] += 0.5 * (abscos(ge1, e1) + abscos(ge2, e2))
# realizable decision-directed SIC: the stronger user is detected
# with the same aware Wiener stage (a scalar-scaled matched-filter
# decision would re-modulate to a multiple of r itself, because
# M_s M_s^T = I, and cancel nothing), its re-modulated estimate is
# subtracted, and the weaker user is read from the residual.
if abs(h1) >= abs(h2):
hs, hw, Ms, Mw, es, ew = h1, h2, M1, M2, e1, e2
Qsw, csw, vsw = Q, c1, v1
else:
hs, hw, Ms, Mw, es, ew = h2, h1, M2, M1, e2, e1
Qsw, csw, vsw = Q.T, c2, v2
t_s = (Ms.T.to(torch.complex64) @ r.unsqueeze(-1)).squeeze(-1) / hs
dec = aware_g(t_s, Qsw, beta, csw, vsw)
r_res = r - hs * (Ms.to(torch.complex64)
@ dec.unsqueeze(-1)).squeeze(-1)
d_w = (Mw.T.to(torch.complex64) @ r_res.unsqueeze(-1)).squeeze(-1) / hw
res["sic"] += 0.5 * (abscos(dec, es) + abscos(d_w, ew))
if (tr + 1) % 100 == 0:
print(f" {tr+1}/{ntr} ({time.time()-t0:.0f}s)", flush=True)
for k in res:
res[k] /= ntr
rows = [[s] + [res[k][i] for k in ("edma", "blind", "oma", "genie", "sic")]
for i, s in enumerate(snr_db)]
write_csv("sic_comparison",
["snr_db", "edma", "blind", "oma", "genie", "sic"], rows)
i20 = list(snr_db).index(20)
print(f" at 20 dB: EDMA {res['edma'][i20]:.3f}, blind "
f"{res['blind'][i20]:.3f}, SIC {res['sic'][i20]:.3f}, "
f"genie {res['genie'][i20]:.3f}, OMA {res['oma'][i20]:.3f}")
# ------------------------------------------------------------------
def E3_unconditional(beta=0.311, d=512, ntr=1500):
print("\n=== E3: Rayleigh unconditional MSE of the aware receiver ===")
rows = []
for s in (10, 20):
sig = 10 ** (-s / 20.0)
mses, bl = [], []
for _ in range(ntr):
e1, e2 = embed_pair(d, beta)
M1, M2 = haar_g(d), haar_g(d)
Q = M1.T @ M2
h1, h2 = rayleigh2()
n = cnoise_g(d)
r = h1 * (M1 @ e1).to(torch.complex64) \
+ h2 * (M2 @ e2).to(torch.complex64) + sig * n
t1 = (M1.T.to(torch.complex64) @ r) / h1
c1 = h2 / h1
nv = torch.tensor([sig**2 / abs(h1)**2], device=DEV)
g1 = aware_g(t1.unsqueeze(0), Q, beta, c1, nv)[0]
mses.append(float((g1 - e1.to(torch.complex64)).norm()**2))
lam = (1.0 / d) / (1.0 / d + abs(c1)**2 / d + sig**2 / abs(h1)**2)
b1 = lam * t1
bl.append(float((b1 - e1.to(torch.complex64)).norm()**2))
rows.append([s, float(np.mean(mses)), float(np.median(mses)),
float(np.mean(bl)), float(np.median(bl))])
print(f" {s} dB: aware mean {rows[-1][1]:.4f} "
f"(median {rows[-1][2]:.4f}) | blind mean {rows[-1][3]:.4f} "
f"(median {rows[-1][4]:.4f})")
write_csv("rayleigh_mse", ["snr_db", "aware_mean", "aware_median",
"blind_mean", "blind_median"], rows)
# ------------------------------------------------------------------
def E4_csi(beta=0.311, d=512, snr=30.0, ntr=400):
print("\n=== E4: imperfect CSI robustness (EDMA vs realizable SIC) ===")
sh2 = np.array([0.0, 0.01, 0.02, 0.05, 0.1, 0.2, 0.3])
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 = haar_g(d), haar_g(d)
Q = M1.T @ M2
h1, h2 = rayleigh2()
n = cnoise_g(d)
r = h1 * (M1 @ e1).to(torch.complex64) \
+ h2 * (M2 @ e2).to(torch.complex64) + sig * n
eps1, eps2 = rayleigh2()
for j, v in enumerate(sh2):
hh1 = h1 + math.sqrt(v) * eps1
hh2 = h2 + math.sqrt(v) * eps2
t1 = (M1.T.to(torch.complex64) @ r) / hh1
t2 = (M2.T.to(torch.complex64) @ r) / hh2
nv1 = torch.tensor([sig**2 / abs(hh1)**2], device=DEV)
nv2 = torch.tensor([sig**2 / abs(hh2)**2], device=DEV)
g1 = aware_g(t1.unsqueeze(0), Q, beta, hh2 / hh1, nv1)[0]
g2 = aware_g(t2.unsqueeze(0), Q.T, beta, hh1 / hh2, nv2)[0]
res_e[j] += 0.5 * (float(abscos(g1.unsqueeze(0), e1)[0])
+ float(abscos(g2.unsqueeze(0), e2)[0]))
if abs(hh1) >= abs(hh2):
hs, hw, Ms, Mw, es, ew = hh1, hh2, M1, M2, e1, e2
Qsw, csw = Q, hh2 / hh1
else:
hs, hw, Ms, Mw, es, ew = hh2, hh1, M2, M1, e2, e1
Qsw, csw = Q.T, hh1 / hh2
t_s = (Ms.T.to(torch.complex64) @ r) / hs
nvs = torch.tensor([sig**2 / abs(hs)**2], device=DEV)
dec = aware_g(t_s.unsqueeze(0), Qsw, beta, csw, nvs)[0]
r_res = r - hs * (Ms.to(torch.complex64) @ dec)
d_w = (Mw.T.to(torch.complex64) @ r_res) / hw
res_s[j] += 0.5 * (float(abscos(dec.unsqueeze(0), es)[0])
+ float(abscos(d_w.unsqueeze(0), ew)[0]))
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)")
rows = [[v, res_e[j], res_s[j]] for j, v in enumerate(sh2)]
write_csv("csi_error", ["sigma_h2", "edma", "sic"], rows)
# ------------------------------------------------------------------
def E5_maskfam(beta=0.311, d=512, ntr=200):
print("\n=== E5: Walsh-Hadamard diagonal variant vs Haar ===")
snr_db = np.arange(0, 41, 5)
sigs = torch.tensor(10 ** (-snr_db / 20.0), dtype=torch.float32,
device=DEV)
nb = len(snr_db)
H = np.array([[1.0]])
while H.shape[0] < d:
H = np.block([[H, H], [H, -H]])
Ht = torch.tensor(H / math.sqrt(d), dtype=torch.float32, device=DEV)
g = 1.0 - beta**2
res = {"haar": np.zeros(nb), "wh": np.zeros(nb)}
exact = np.zeros(nb)
for _ in range(ntr):
e1, e2 = embed_pair(d, beta)
M1, M2 = haar_g(d), haar_g(d)
Q = M1.T @ M2
D1 = torch.tensor(np.sign(rng.standard_normal(d)),
dtype=torch.float32, device=DEV)
D2 = torch.tensor(np.sign(rng.standard_normal(d)),
dtype=torch.float32, device=DEV)
W1, W2 = Ht * D1.unsqueeze(0), Ht * D2.unsqueeze(0)
q = D1 * D2
n = cnoise_g(d)
r0h = (M1 @ e1 + M2 @ e2).to(torch.complex64)
r0w = (W1 @ e1 + W2 @ e2).to(torch.complex64)
rh = r0h.unsqueeze(0) + sigs.view(-1, 1) * n.unsqueeze(0)
rw = r0w.unsqueeze(0) + sigs.view(-1, 1) * n.unsqueeze(0)
t1 = (M1.T.to(torch.complex64) @ rh.unsqueeze(-1)).squeeze(-1)
g1 = aware_g(t1, Q, beta, 1.0, sigs**2)
res["haar"] += abscos(g1, e1)
tw = (W1.T.to(torch.complex64) @ rw.unsqueeze(-1)).squeeze(-1)
a = 1.0 + beta * q # (d,)
rho = g / d + sigs**2 # (nb,)
wdiag = a.unsqueeze(0) / (a.unsqueeze(0)**2 / d
+ rho.view(-1, 1)) # (nb,d)
w1 = (wdiag.to(torch.complex64) / d) * tw
res["wh"] += abscos(w1, e1)
exact += ((g + d * sigs.view(-1, 1)**2)
/ (a.unsqueeze(0)**2 + g + d * sigs.view(-1, 1)**2)
).mean(1).cpu().numpy()
for k in res:
res[k] /= ntr
exact /= ntr
print(f" max |WH - Haar| cosine dev: "
f"{100*np.max(np.abs(res['wh']-res['haar'])):.2f} points; "
f"40 dB WH {res['wh'][-1]:.4f} vs Haar {res['haar'][-1]:.4f}")
rows = [[s, res["haar"][i], res["wh"][i], exact[i]]
for i, s in enumerate(snr_db)]
write_csv("mask_family_rev", ["snr_db", "haar", "wh", "wh_exact_mse"],
rows)
# ------------------------------------------------------------------
def E7_multiuser(beta=0.311, d=512, ntr=100):
print("\n=== E7c: multi-user scaling (joint Wiener) ===")
snr_db = np.arange(0, 31, 2.5)
sigs = torch.tensor(10 ** (-snr_db / 20.0), dtype=torch.float32,
device=DEV)
nb = len(snr_db)
rows = []
for U in (2, 3, 4):
B = (1 - beta) * np.eye(U) + beta * np.ones((U, U))
A = np.linalg.cholesky(B)
Bt = torch.tensor(B, dtype=torch.float32, device=DEV)
err = np.zeros(nb)
t0 = time.time()
for _ in range(ntr):
F = rng.standard_normal((d, U))
Fq, _ = np.linalg.qr(F)
E = (Fq @ A.T).T
Et = torch.tensor(E, dtype=torch.float32, device=DEV)
masks = [haar_g(d) for _ in range(U)]
n = cnoise_g(d)
r0 = sum(masks[u] @ Et[u] for u in range(U)).to(torch.complex64)
Qs = [masks[0].T @ masks[v] for v in range(U)]
Ret = sum(Bt[0, v] * Qs[v].T for v in range(U)) / d
S0 = sum(Bt[v, w] * (Qs[v] @ Qs[w].T)
for v in range(U) for w in range(U)) / d
r = r0.unsqueeze(0) + sigs.view(-1, 1) * n.unsqueeze(0)
t = (masks[0].T.to(torch.complex64)
@ r.unsqueeze(-1)).squeeze(-1)
S = S0.to(torch.complex64).unsqueeze(0) \
+ (sigs**2).view(-1, 1, 1) \
* torch.eye(d, device=DEV, dtype=torch.complex64)
x = torch.linalg.solve(S, t.unsqueeze(-1))
eh = (Ret.to(torch.complex64).unsqueeze(0) @ x).squeeze(-1)
err += ((eh - Et[0].to(torch.complex64)).norm(dim=1)**2
).cpu().numpy()
err /= ntr
T_mc = U * np.log2(1.0 / err)
r0v = (U - 1) + d / 10 ** (snr_db / 10.0)
T_bl = U * np.log2(1.0 + 1.0 / r0v)
T_oma = U * np.log2(1 + 10 ** (snr_db / 10.0) / (U * d))
for i, s in enumerate(snr_db):
rows.append([U, s, T_mc[i], T_bl[i], T_oma[i], err[i]])
i20 = list(snr_db).index(20.0)
print(f" U={U}: at 20 dB EDMA {T_mc[i20]:.3f} vs blind "
f"{T_bl[i20]:.3f} vs OMA {T_oma[i20]:.3f} "
f"(gain {T_mc[i20]/T_oma[i20]:.2f}x), floor MSE {err[-1]:.4f}"
f" [{time.time()-t0:.0f}s]")
write_csv("multiuser_corrected",
["U", "snr_db", "edma_mc", "blind", "oma", "mse_mc"], rows)
# ------------------------------------------------------------------
def E8_mismatch(beta=0.311, d=512, snr=20.0, ntr=200):
print("\n=== E8: affinity mismatch of the aware receiver ===")
sig = 10 ** (-snr / 20.0)
bhs = [b for b in (beta - 0.1, beta - 0.06, beta, beta + 0.06,
beta + 0.1, beta + 2 ** -8, 0.0) if b >= 0]
accs = np.zeros(len(bhs)); msea = np.zeros(len(bhs))
for _ in range(ntr):
e1, e2 = embed_pair(d, beta)
M1, M2 = haar_g(d), haar_g(d)
Q = M1.T @ M2
n = cnoise_g(d)
r = (M1 @ e1 + M2 @ e2).to(torch.complex64) + sig * n
t1 = (M1.T.to(torch.complex64) @ r)
nv = torch.tensor([sig**2], device=DEV)
for j, bh in enumerate(bhs):
g1 = aware_g(t1.unsqueeze(0), Q, bh, 1.0, nv)[0]
accs[j] += float(abscos(g1.unsqueeze(0), e1)[0])
msea[j] += float((g1 - e1.to(torch.complex64)).norm()**2)
accs /= ntr; msea /= ntr
rows = []
for j, bh in enumerate(bhs):
tag = ("quant b=7" if abs(bh - beta - 2**-8) < 1e-12 else
("blind" if bh == 0.0 else f"delta={bh-beta:+.2f}"))
print(f" beta_hat={bh:.4f} ({tag}): cosine {accs[j]:.4f}, "
f"MSE {msea[j]:.4f}")
rows.append([bh, accs[j], msea[j]])
write_csv("mismatch", ["beta_hat", "cosine", "mse"], rows)
# ------------------------------------------------------------------
def E9_ceiling(d=512, snr=60.0, ntr=200):
print(chr(10) + '=== E9: cosine-ceiling verification (h=1) ===')
sig = 10 ** (-snr / 20.0)
rows = []
for beta in (0.311, 0.8):
g = 1.0 - beta**2
acc_a = 0.0; acc_b = 0.0
for _ in range(ntr):
e1, e2 = embed_pair(d, beta)
M1, M2 = haar_g(d), haar_g(d)
Q = M1.T @ M2
n = cnoise_g(d)
r = (M1 @ e1 + M2 @ e2).to(torch.complex64) + sig * n
t1 = (M1.T.to(torch.complex64) @ r)
nv = torch.tensor([sig**2], device=DEV)
g1 = aware_g(t1.unsqueeze(0), Q, beta, 1.0, nv)[0]
acc_a += float(abscos(g1.unsqueeze(0), e1)[0])
acc_b += float(abscos(t1.unsqueeze(0), e1)[0])
acc_a /= ntr; acc_b /= ntr
import math as _m
pred_a = _m.sqrt(1.0 - _m.sqrt(g) / 2.0)
pred_b = _m.sqrt(0.5)
print(f' beta={beta}: aware MC {acc_a:.4f} pred {pred_a:.4f} | '
f'blind MC {acc_b:.4f} pred {pred_b:.4f}')
rows.append([beta, acc_a, pred_a, acc_b, pred_b])
write_csv('cosine_ceiling', ['beta', 'aware_mc', 'aware_pred',
'blind_mc', 'blind_pred'], rows)
if __name__ == "__main__":
todo = set(sys.argv[1:])
ALL = {"E2": E2_sic, "E3": E3_unconditional, "E4": E4_csi,
"E5": E5_maskfam, "E7c": E7_multiuser, "E8": E8_mismatch,
"E9": E9_ceiling}
for name, fn in ALL.items():
if not todo or name in todo:
fn()
print("\nAll requested GPU simulations complete.")
+225 -338
View File
@@ -1,70 +1,35 @@
"""
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.
Monte Carlo verification of every closed-form claim (v2 design).
================================================================
Independent Haar masks + affinity-aware Wiener demultiplexer.
Checks (d = 256 for speed; deviations shrink as O(1/d)):
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
V1 Theorem 1 MSE formula vs MC at several (beta, SNR), h = 1
V2 Theorem 1 under random channel phases, both users
V3 floors: aware sqrt(g)/2 vs blind 1/2, and the value ratio
V4 cosine ceiling sqrt(1 - MSE) (Corollary: cosine)
V5 blind receiver == matched filter in cosine (scalar shrinkage)
V6 monotonicity of the MSE in beta (Proposition)
V7 full-cooperation bound T <= log2(1+4 rho/d), equality at beta=1
V8 MAC condition gamma^2 (2+k) >= 2 beta^2 k^2 boundary
V9 Walsh-Hadamard diagonal variant: exact finite-d closed form
V10 mismatch stationarity: MSE(beta_hat) - MSE(beta) = O(delta^2)
V11 correlated-mask alternative floor 1 + 4 beta^4 / gamma
(Remark and Appendix), dominated by the aware receiver
Pure numpy, fixed seed, ~2 minutes on a laptop.
"""
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}")
D = 256
def haar(d):
Q, R = np.linalg.qr(rng.standard_normal((d, d)))
G = rng.standard_normal((d, d))
Q, R = np.linalg.qr(G)
return Q * np.sign(np.diag(R))
@@ -72,303 +37,225 @@ def unit(v):
return v / np.linalg.norm(v)
def pair(d, beta):
def cosim(a, b):
return float(abs(np.vdot(a, b)) / (np.linalg.norm(a) * np.linalg.norm(b)))
def embed_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
return e1, beta * e1 + math.sqrt(1 - beta * beta) * 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 cnoise(d):
return (rng.standard_normal(d) + 1j * rng.standard_normal(d)) / math.sqrt(2)
def csi2(beta, c):
g = 1 - beta**2
return (abs(c)**2 + beta**2 + 2 * beta**2 * np.real(c)) / g
def mse_theory(beta, c1, rho_e):
a0 = 1.0 + beta**2 * abs(c1)**2 + rho_e
return rho_e / math.sqrt(a0 * a0 - 4.0 * beta**2 * abs(c1)**2)
# ---------------- 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}")
def aware(t1, Q, beta, c1, nvar, d):
g = 1.0 - beta * beta
rho = g * abs(c1)**2 / d + nvar
S = beta * (c1 * Q + np.conj(c1) * Q.T) / d
S[np.diag_indices(d)] += (1.0 + beta**2 * abs(c1)**2) / d + rho
x = np.linalg.solve(S, t1)
return (x + beta * np.conj(c1) * (Q.T @ x)) / d
# ---------------- 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}%")
def run_pair(beta, sig, h1=1.0 + 0j, h2=1.0 + 0j, d=D):
e1, e2 = embed_pair(d, beta)
M1, M2 = haar(d), haar(d)
Q = M1.T @ M2
r = h1 * (M1 @ e1) + h2 * (M2 @ e2) + sig * cnoise(d)
t1 = M1.T @ r / h1
return e1, e2, Q, t1, M2.T @ r / h2
# ---------------- V4: quoted constants -----------------------------
def check(name, ok, detail=""):
print(f"[{'PASS' if ok else 'FAIL'}] {name} {detail}")
return ok
allok = True
# ---------------------------------------------------------------- V1
devs = []
for beta in (0.0, 0.311, 0.6):
for snr in (10.0, 20.0, 60.0):
sig = 10 ** (-snr / 20.0)
mc = 0.0
NT = 40
for _ in range(NT):
e1, _, Q, t1, _ = run_pair(beta, sig)
g1 = aware(t1, Q, beta, 1.0, sig * sig, D)
mc += float(np.linalg.norm(g1 - e1) ** 2)
mc /= NT
th = mse_theory(beta, 1.0, (1 - beta**2) + D * sig * sig)
devs.append(abs(mc / th - 1))
allok &= check("V1 Theorem 1 (h=1)", max(devs) < 0.03,
f"max dev {100*max(devs):.2f}%")
# ---------------------------------------------------------------- V2
devs = []
sig = 10 ** (-20.0 / 20.0)
for beta in (0.311, 0.5):
for _ in range(30):
h1 = np.exp(1j * rng.uniform(0, 2 * np.pi))
h2 = np.exp(1j * rng.uniform(0, 2 * np.pi))
e1, e2, Q, t1, t2 = run_pair(beta, sig, h1, h2)
c1, c2 = h2 / h1, h1 / h2
g1 = aware(t1, Q, beta, c1, sig**2, D)
g2 = aware(t2, Q.T, beta, c2, sig**2, D)
g = 1 - beta**2
th1 = mse_theory(beta, c1, g * abs(c1)**2 + D * sig**2)
th2 = mse_theory(beta, c2, g * abs(c2)**2 + D * sig**2)
devs.append(abs(np.linalg.norm(g1 - e1)**2 / th1 - 1))
devs.append(abs(np.linalg.norm(g2 - e2)**2 / th2 - 1))
allok &= check("V2 Theorem 1 (random phases, both users)",
float(np.mean(devs)) < 0.05,
f"mean dev {100*float(np.mean(devs)):.2f}%")
# ---------------------------------------------------------------- V3
beta = 0.6
sig = 1e-3
mc_a, mc_b = 0.0, 0.0
NT = 40
for _ in range(NT):
e1, _, Q, t1, _ = run_pair(beta, sig)
g1 = aware(t1, Q, beta, 1.0, sig * sig, D)
mc_a += float(np.linalg.norm(g1 - e1) ** 2)
lam = (1.0 / D) / (2.0 / D + sig * sig)
mc_b += float(np.linalg.norm(lam * t1 - e1) ** 2)
mc_a /= NT
mc_b /= NT
fa, fb = math.sqrt(1 - beta**2) / 2, 0.5
allok &= check("V3 floors sqrt(g)/2 vs 1/2",
abs(mc_a - fa) < 0.02 and abs(mc_b - fb) < 0.02,
f"aware {mc_a:.4f}~{fa:.4f}, blind {mc_b:.4f}~{fb:.4f}, "
f"ratio {mc_b/mc_a:.3f}~{1/math.sqrt(1-beta**2):.3f}")
# ---------------------------------------------------------------- V4
acc = 0.0
for _ in range(NT):
e1, _, Q, t1, _ = run_pair(beta, sig)
acc += cosim(aware(t1, Q, beta, 1.0, sig * sig, D), e1)
acc /= NT
pred = math.sqrt(1 - fa)
allok &= check("V4 cosine ceiling sqrt(1-MSE)", abs(acc - pred) < 0.01,
f"MC {acc:.4f} vs {pred:.4f}")
# ---------------------------------------------------------------- V5
e1, _, Q, t1, _ = run_pair(0.311, 0.1)
lam = 0.37 # any scalar
allok &= check("V5 blind == MF in cosine",
abs(cosim(lam * t1, e1) - cosim(t1, e1)) < 1e-12)
# ---------------------------------------------------------------- V6
k = D / 100.0
vals = [mse_theory(b, 1.0, (1 - b * b) + k)
for b in np.linspace(0, 0.99, 50)]
allok &= check("V6 monotonic decrease in beta",
all(x > y for x, y in zip(vals, vals[1:])))
# ---------------------------------------------------------------- V7
ok7 = True
worst = 0.0
for rho in (1.0, 100.0, 1e4):
kk = D / rho
coop = math.log2(1 + 4 * rho / D)
for b in np.linspace(0, 1.0, 41):
m = mse_theory(b, 1.0, (1 - b * b) + kk)
T = 2 * math.log2(1 / m)
ok7 &= T <= coop + 1e-9
worst = max(worst, T - coop)
m1 = mse_theory(1.0, 1.0, kk)
ok7 &= abs(2 * math.log2(1 / m1) - coop) < 1e-9
allok &= check("V7 full-cooperation bound, equality at beta=1", ok7,
f"max T-coop {worst:.2e}")
# ---------------------------------------------------------------- V8
ok8 = True
for rho in (1.0, 10.0, 100.0, 1e3):
kk = D / rho
for b in (0.1, 0.311, 0.6, 0.9):
g = 1 - b * b
m = mse_theory(b, 1.0, g + kk)
T = 2 * math.log2(1 / m)
mac = math.log2(1 + 2 * rho / D)
lhs = g * g * (2 + kk)
rhs = 2 * b * b * kk * kk
ok8 &= (T <= mac + 1e-9) == (lhs >= rhs - 1e-9)
allok &= check("V8 MAC-condition boundary", ok8)
# ---------------------------------------------------------------- V9
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)")
sig = 10 ** (-20.0 / 20.0)
H = np.array([[1.0]])
while H.shape[0] < D:
H = np.block([[H, H], [H, -H]])
H /= math.sqrt(D)
mc, th = 0.0, 0.0
for _ in range(30):
e1, e2 = embed_pair(D, beta)
D1 = np.sign(rng.standard_normal(D))
D2 = np.sign(rng.standard_normal(D))
W1, W2 = H * D1[None, :], H * D2[None, :]
r = W1 @ e1 + W2 @ e2 + sig * cnoise(D)
t1 = W1.T @ r
q = D1 * D2
a = 1.0 + beta * q
rho = g / D + sig * sig
w1 = (a / (a * a / D + rho)) * t1 / D
mc += float(np.linalg.norm(w1 - e1) ** 2)
th += float(np.mean((g + D * sig**2) / (a * a + g + D * sig**2)))
allok &= check("V9 WH exact finite-d closed form",
abs(mc / th - 1) < 0.03, f"dev {100*abs(mc/th-1):.2f}%")
# ---------------- 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)
# ---------------------------------------------------------------- V10
beta = 0.3
sig = 10 ** (-20.0 / 20.0)
base, d1, d2 = 0.0, 0.0, 0.0
for _ in range(30):
e1, _, Q, t1, _ = run_pair(beta, sig)
for bh, tag in ((beta, "b"), (beta + 0.2, "1"), (beta + 0.4, "2")):
g1 = aware(t1, Q, bh, 1.0, sig * sig, D)
m = float(np.linalg.norm(g1 - e1) ** 2)
if tag == "b":
base += m
elif tag == "1":
d1 += m
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})")
d2 += m
base /= 30; d1 /= 30; d2 /= 30
r_quad = (d2 - base) / max(d1 - base, 1e-12)
allok &= check("V10 quadratic mismatch (delta doubling ~ 4x)",
2.5 < r_quad < 6.5,
f"MSE(+0)={base:.4f} MSE(+0.2)={d1:.4f} "
f"MSE(+0.4)={d2:.4f} ratio {r_quad:.2f}")
# ---------------- 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}")
# ---------------------------------------------------------------- V11
beta = 0.311
g = 1 - beta**2
mc = 0.0
for _ in range(30):
e1, e2 = embed_pair(D, beta)
U1, U2 = haar(D), haar(D)
M1, M2 = U1, beta * U1 + math.sqrt(g) * U2
r = M1 @ e1 + M2 @ e2 # noise-free -> floor
t1 = M1.T @ r
t2 = M2.T @ r
g1 = (t1 - beta * t2) / g
mc += float(np.linalg.norm(g1 - e1) ** 2)
mc /= 30
th = 1 + 4 * beta**4 / g
allok &= check("V11 correlated-mask floor 1+4b^4/g",
abs(mc / th - 1) < 0.05,
f"MC {mc:.4f} vs {th:.4f}; aware floor "
f"{math.sqrt(g)/2:.4f} (dominated)")
print()
print("=" * 60)
print(f"RESULT: {'ALL PASS' if not FAIL else 'FAILURES: ' + ', '.join(FAIL)}")
print("\nALL CHECKS PASSED" if allok else "\nSOME CHECKS FAILED")