""" 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.")