198 lines
8.8 KiB
Python
Executable File
198 lines
8.8 KiB
Python
Executable File
"""
|
|
Auxiliary experiments for the TWC revision (main_FFF.tex).
|
|
Reuses the validated simulation primitives in semantic_correlation_sim.py to
|
|
produce REAL numbers for the new reviewer-requested studies:
|
|
|
|
A. Threshold sensitivity (tau = 0.30..0.50) + mean cosine similarity [R1.2, R3.2]
|
|
B. Residual phase-error robustness [R1.1, R2.3]
|
|
C. User scaling U in {4,8,16,32} : full vs. sparse top-k attention [R1.6, R3.4]
|
|
D. DL semantic baseline: Joint+Attn calibrated at a single nominal SNR [R1.5,R2.5,R3.5]
|
|
vs. MAML SNR-adaptive shrinkage
|
|
|
|
All results are printed as LaTeX-ready rows.
|
|
"""
|
|
import numpy as np
|
|
import semantic_correlation_sim as sim
|
|
|
|
RNG = np.random.default_rng(2026)
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# A. Threshold sensitivity + mean cosine similarity (HIGH/LOW/MIX, SNR=10 dB)
|
|
# ----------------------------------------------------------------------------
|
|
def collect_cos(scenario_key, snr=10.0, n_mc=600):
|
|
cfg = sim.SCENARIOS[scenario_key]
|
|
beta_mat = sim.compute_beta_matrix(cfg)
|
|
out = {'OFDMA': [], 'NOMA-SIC': [], 'UWCA': []}
|
|
for _ in range(n_mc):
|
|
Egt = sim.gen_embeddings(sim.BATCH, scenario_key)
|
|
Y = sim.shared_embedding_channel(Egt, snr)
|
|
Eh, _ = sim.ofdma_se_decoder(Y)
|
|
out['OFDMA'].append(sim.cos_sim(Eh, Egt).ravel())
|
|
y, h = sim.noma_ul_channel(Egt, snr)
|
|
Eh = sim.noma_sic_decoder(y, h)
|
|
out['NOMA-SIC'].append(sim.cos_sim(Eh, Egt).ravel())
|
|
Y = sim.shared_embedding_channel(Egt, snr)
|
|
Eh, _ = sim.maml_attention_se_decoder(Y, snr, beta_mat)
|
|
out['UWCA'].append(sim.cos_sim(Eh, Egt).ravel())
|
|
return {k: np.concatenate(v) for k, v in out.items()}
|
|
|
|
def exp_A():
|
|
print("\n=== EXP A: threshold sensitivity + mean cosine (SNR=10 dB) ===")
|
|
taus = [0.30, 0.35, 0.40, 0.45, 0.50]
|
|
for scen in ['HIGH', 'LOW', 'MIX']:
|
|
cos = collect_cos(scen)
|
|
print(f"\n[{scen}]")
|
|
for m in ['OFDMA', 'NOMA-SIC', 'UWCA']:
|
|
c = cos[m]
|
|
sers = [f"{(c < t).mean():.3f}" for t in taus]
|
|
print(f" {m:9s} meancos={c.mean():.3f} SER@tau[{','.join(map(str,taus))}] = {sers}")
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# B. Residual phase-error robustness
|
|
# After imperfect pilot-based compensation, residual phase Dphi ~ N(0,sig^2);
|
|
# recovered in-phase component scales by cos(Dphi) (quadrature energy lost).
|
|
# ----------------------------------------------------------------------------
|
|
def se_channel_phase(E, snr_db, sigma_phi_deg):
|
|
n, U, D = E.shape
|
|
X = E * sim.MASKS[None, :, :]
|
|
Ytx = X.sum(axis=1)
|
|
h = (np.sqrt(RNG.standard_normal((n, U, 1))**2 + RNG.standard_normal((n, U, 1))**2)
|
|
* np.sqrt(0.5))
|
|
sig_power = float(np.mean(Ytx**2))
|
|
noise_std = np.sqrt(sig_power / (10**(snr_db/10)))
|
|
phi = np.deg2rad(sigma_phi_deg) * RNG.standard_normal((n, U, 1))
|
|
Yrx = h * np.cos(phi) * Ytx[:, None, :] + RNG.standard_normal((n, U, D)) * noise_std
|
|
return Yrx
|
|
|
|
def exp_B():
|
|
print("\n=== EXP B: residual phase-error robustness (HIGH scenario) ===")
|
|
cfg = sim.SCENARIOS['HIGH']; beta_mat = sim.compute_beta_matrix(cfg)
|
|
for snr in [10.0, 20.0]:
|
|
row = []
|
|
for sig in [0, 5, 10, 15, 20]:
|
|
acc = 0.0; n_mc = 400
|
|
for _ in range(n_mc):
|
|
Egt = sim.gen_embeddings(sim.BATCH, 'HIGH')
|
|
Y = se_channel_phase(Egt, snr, sig)
|
|
Eh, _ = sim.maml_attention_se_decoder(Y, snr, beta_mat)
|
|
acc += sim.ser_total(Eh, Egt)
|
|
row.append(f"{acc/n_mc:.3f}")
|
|
print(f" SNR={snr:4.0f}dB UWCA-SER vs sigma_phi[0,5,10,15,20 deg] = {row}")
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# C. User scaling + sparse top-k attention (clustered relevance)
|
|
# U users in clusters of size g sharing a scene; cross-cluster beta=0.
|
|
# full attention: O(U^2) ; top-k (k=g): O(U*k).
|
|
# ----------------------------------------------------------------------------
|
|
def gen_clustered(n, U, D, g, beta):
|
|
n_clusters = U // g
|
|
scenes = []
|
|
for _ in range(n_clusters):
|
|
s = RNG.standard_normal(D); scenes.append(s / np.linalg.norm(s))
|
|
embs = []
|
|
for u in range(U):
|
|
s = scenes[u // g]
|
|
priv = RNG.standard_normal((n, D))
|
|
priv /= np.linalg.norm(priv, axis=-1, keepdims=True) + 1e-8
|
|
e = np.sqrt(1 - beta**2) * priv + beta * s[None, :]
|
|
e /= np.linalg.norm(e, axis=-1, keepdims=True) + 1e-8
|
|
embs.append(e)
|
|
return np.stack(embs, axis=1)
|
|
|
|
def masks_for(U, D):
|
|
dpu = D // U
|
|
M = np.zeros((U, D))
|
|
for u in range(U):
|
|
M[u, u*dpu:(u+1)*dpu] = 1.0
|
|
return M
|
|
|
|
def se_channel_generic(E, snr_db, M):
|
|
n, U, D = E.shape
|
|
X = E * M[None, :, :]
|
|
Ytx = X.sum(axis=1)
|
|
h = (np.sqrt(RNG.standard_normal((n, U, 1))**2 + RNG.standard_normal((n, U, 1))**2)
|
|
* np.sqrt(0.5))
|
|
noise_std = np.sqrt(float(np.mean(Ytx**2)) / (10**(snr_db/10)))
|
|
return h * Ytx[:, None, :] + RNG.standard_normal((n, U, D)) * noise_std
|
|
|
|
def attn_decode(Yrx, M, beta_mat, topk=None):
|
|
n, U, D = Yrx.shape
|
|
R = Yrx[:, :, None, :] * M[None, None, :, :] # (n,U,U,D)
|
|
alpha = beta_mat.copy(); np.fill_diagonal(alpha, 1.0)
|
|
if topk is not None and topk < U:
|
|
# keep self + top-(k-1) strongest cross weights per row
|
|
for u in range(U):
|
|
order = np.argsort(-alpha[u])
|
|
keep = set(order[:topk].tolist()) | {u}
|
|
for v in range(U):
|
|
if v not in keep:
|
|
alpha[u, v] = 0.0
|
|
alpha /= alpha.sum(1, keepdims=True) + 1e-8
|
|
ctx = np.einsum('ui,buid->bud', alpha, R)
|
|
Eh = np.stack([sim._norm(ctx[:, u, :]) for u in range(U)], axis=1)
|
|
return Eh
|
|
|
|
def exp_C():
|
|
print("\n=== EXP C: user scaling + sparse top-k attention (g=4, beta=0.65, SNR=10 dB) ===")
|
|
g = 4; beta = 0.65; snr = 10.0
|
|
for U in [4, 8, 16, 32]:
|
|
D = 16 * U # keep 16 dims/user
|
|
M = masks_for(U, D)
|
|
bm = np.zeros((U, U))
|
|
for i in range(U):
|
|
for j in range(U):
|
|
if i // g == j // g:
|
|
bm[i, j] = beta * beta
|
|
n_mc = 200
|
|
ser_full = ser_topk = 0.0
|
|
for _ in range(n_mc):
|
|
Egt = gen_clustered(sim.BATCH, U, D, g, beta)
|
|
Y = se_channel_generic(Egt, snr, M)
|
|
ser_full += float((sim.cos_sim(attn_decode(Y, M, bm), Egt) < 0.45).mean())
|
|
Y2 = se_channel_generic(Egt, snr, M)
|
|
ser_topk += float((sim.cos_sim(attn_decode(Y2, M, bm, topk=g), Egt) < 0.45).mean())
|
|
ops_full = U * U
|
|
ops_topk = U * g
|
|
print(f" U={U:3d} SER_full={ser_full/n_mc:.3f} SER_topk(k={g})={ser_topk/n_mc:.3f}"
|
|
f" attn_ops: full={ops_full} topk={ops_topk} reduction={ops_full/ops_topk:.1f}x")
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# D. DL semantic baseline: Joint+Attn calibrated at single nominal SNR (10 dB)
|
|
# vs. MAML SNR-adaptive shrinkage. Wiener-type shrinkage s = g/(g+1) applied
|
|
# to the aggregated cross-attention context; MAML adapts s to the test SNR,
|
|
# the non-meta Joint baseline is frozen at the training SNR.
|
|
# ----------------------------------------------------------------------------
|
|
def attn_decode_shrink(Yrx, M, beta_mat, shrink):
|
|
n, U, D = Yrx.shape
|
|
R = Yrx[:, :, None, :] * M[None, None, :, :]
|
|
alpha = beta_mat.copy(); np.fill_diagonal(alpha, 0.0)
|
|
alpha /= (alpha.sum(1, keepdims=True) + 1e-8)
|
|
cross = np.einsum('ui,buid->bud', alpha, R) # cross context
|
|
own = np.einsum('buud->bud', R.transpose(0,1,2,3)) # placeholder
|
|
own = Yrx * M[None, :, :] # own subspace skip
|
|
ctx = shrink * cross + own
|
|
Eh = np.stack([sim._norm(ctx[:, u, :]) for u in range(U)], axis=1)
|
|
return Eh
|
|
|
|
def exp_D():
|
|
print("\n=== EXP D: DL baseline (Joint+Attn fixed 10 dB) vs MAML adaptive ===")
|
|
cfg = sim.SCENARIOS['HIGH']; bm = sim.compute_beta_matrix(cfg)
|
|
g0 = 10**(10/10); shrink_fixed = g0/(g0+1) # calibrated at 10 dB
|
|
for snr in [0, 5, 10, 15, 20]:
|
|
g = 10**(snr/10); shrink_adapt = g/(g+1)
|
|
n_mc = 300; ser_fixed = ser_adapt = 0.0
|
|
for _ in range(n_mc):
|
|
Egt = sim.gen_embeddings(sim.BATCH, 'HIGH')
|
|
Y = sim.shared_embedding_channel(Egt, snr)
|
|
ser_fixed += float((sim.cos_sim(attn_decode_shrink(Y, sim.MASKS, bm, shrink_fixed), Egt) < 0.45).mean())
|
|
Y2 = sim.shared_embedding_channel(Egt, snr)
|
|
ser_adapt += float((sim.cos_sim(attn_decode_shrink(Y2, sim.MASKS, bm, shrink_adapt), Egt) < 0.45).mean())
|
|
print(f" SNR={snr:3d}dB Joint+Attn(fixed10dB)={ser_fixed/n_mc:.3f} MAML-UWCA(adaptive)={ser_adapt/n_mc:.3f}")
|
|
|
|
if __name__ == '__main__':
|
|
exp_A()
|
|
exp_B()
|
|
exp_C()
|
|
exp_D()
|
|
print("\nDONE.")
|