""" ============================================================================= Semantic-Correlation-Aware Multi-User Communication Simulation IEEE TCOM: "Exploiting Inter-User Semantic Relevance via Meta-Learned Cross-Attention for Multi-User Wireless Systems" SE (Shared Embedding) Framework --------------------------------- Each user u encodes a source into a D-dimensional embedding e_u (unit-norm). Masking: x_u = e_u ⊙ m_u (user u uses only D/U = DPU dims) TX: y_tx = Σ_u x_u (superimposed signal, full D dims) RX(user u): y_rx,u = h_u · y_tx + n_u (independent Rayleigh per user) OFDMA-SE: ê_u = normalize(y_rx,u ⊙ m_u) — own DPU-dim block only UWCA-SE: ê_u = normalize(Σ_i α_{u,i}·(y_rx,u ⊙ m_i)) — all D dims via cross-attn Semantic relevance model (paper Eq. 2) ----------------------------------------- e_u = sqrt(1 - beta_u^2) * p_hat_u + beta_u * s s : unit-norm shared scene vector p_hat_u : unit-norm private component (independent across users) beta_u : scene contribution fraction in [0, 1] beta_uv = beta_u * beta_v -> inter-user semantic relevance (same scene only) Mutual Information Analysis (Proposition 2) ---------------------------------------------- I_OFDMA-SE = (D/U) · log2(1 + SNR_lin/D) I_UWCA-SE = (D/U) · log2(1 + SNR_lin/D) [own block] + (U-1)(D/U) · log2(1 + β²·SNR_lin / (D + (1-β²)·SNR_lin)) [cross blocks] MI ratio (high SNR): I_UWCA / I_OFDMA → U · β² For U=4, β=0.95: ratio → 4 × 0.9025 = 3.61× Experiments ----------- Exp 1 : SER vs SNR x 5 scenarios (HIGH / LOW / MIX / HETERO / ASYM) Exp 2 : SER gain vs semantic relevance coefficient beta (monotone validation) Exp 3 : Attention weight heat-maps (selective weighting by scenario) Exp 4 : Inter-user correlation rho (decoded embedding quality) Exp 5 : MAML inner-loop steps S ablation Exp 6 : U-user scaling (SER vs SNR for U=1,2,3,4) Exp 7 : Mutual Information vs SNR (analytical bounds, multi-U) Metrics ------- SER : fraction of users with decoded embedding cosine similarity < tau rho_off: mean absolute off-diagonal Pearson correlation of decoded embeddings MI : analytical mutual information bound (bits per channel use per user) ============================================================================= """ import warnings warnings.filterwarnings('ignore') import os import json import numpy as np from scipy.special import exp1 # Exponential integral E1(x) = ∫_x^∞ e^{-t}/t dt # ── Output directories ──────────────────────────────────────────────────────── OUT_DIR = 'results' os.makedirs(OUT_DIR, exist_ok=True) os.makedirs(f'{OUT_DIR}/data', exist_ok=True) # ══════════════════════════════════════════════════════════════════════════════ # 0. Hyperparameters # ══════════════════════════════════════════════════════════════════════════════ RNG = np.random.default_rng(42) D = 64 # embedding dimension U = 4 # number of users TAU = 0.45 # SER cosine-similarity threshold # NOTE: TAU=0.45 chosen so OFDMA-SE (ceiling cos_sim=sqrt(1/U)=0.5 for U=4) # can reach SER→0 at high SNR. TAU=0.85 would give SER=1 always for OFDMA-SE. N_MC = 500 # Monte Carlo trials per SNR point BATCH = 64 # batch size per trial SNR_DB = np.arange(0, 22, 2) # 0..20 dB, step 2 # Orthogonal subspace masks (SE framework): user u uses dims [u*DPU : (u+1)*DPU] DPU = D // U # dimensions per user (64 / 4 = 16) MASKS = np.zeros((U, D)) for _u in range(U): MASKS[_u, _u * DPU : (_u + 1) * DPU] = 1.0 # NOMA power allocation (descending, sums to 1.0) NOMA_POWER = np.array([0.40, 0.30, 0.20, 0.10]) def load_trained_results(scenario_key: str) -> dict: """Load decoder-only trained SER results from maml_semantic.py JSON export. Returns dict with 'snr_db', 'maml_ser', 'joint_ser' arrays, or None if not found.""" path = os.path.join(OUT_DIR, f"trained_{scenario_key}.json") if not os.path.isfile(path): return None with open(path) as f: d = json.load(f) return {k: np.array(v) if isinstance(v, list) else v for k, v in d.items()} # ══════════════════════════════════════════════════════════════════════════════ # 1. Scenario definitions # ══════════════════════════════════════════════════════════════════════════════ # beta_u : scene contribution fraction per user # beta_uv = beta_u * beta_v -> pairwise semantic relevance coefficient # scene_key: scene identifier (same key = shared latent vector) SCENARIOS = { # ------------------------------------------------------------------ # HIGH: All 4 users observe the same intersection scene # beta_uv = 0.95^2 = 0.90 for all pairs -> maximum semantic gain # ------------------------------------------------------------------ 'HIGH': { 'title': 'HIGH Scenario (All Users Correlated)', 'users': ['TL-Camera (U1)', 'Autovehicle (U2)', 'Pedestrian (U3)', 'Queue-Est. (U4)'], 'beta_u': [0.65, 0.65, 0.60, 0.60], 'scenes': ['traffic', 'traffic', 'traffic', 'traffic'], 'color': '#1565C0', }, # ------------------------------------------------------------------ # LOW: Users observe completely different, unrelated contexts # beta_uv ≈ 0 for all cross-pairs (different scenes) # beta_12 = 0.65*0.05 = 0.033, beta_23 = beta_34 ≈ 0.003 # ------------------------------------------------------------------ 'LOW': { 'title': 'LOW Scenario (All Users Uncorrelated)', 'users': ['TL-Camera (U1)', 'TV Viewer (U2)', 'Music Stream (U3)', 'IoT Weather (U4)'], 'beta_u': [0.65, 0.05, 0.05, 0.05], 'scenes': ['traffic', 'home', 'office', 'outdoor'], 'color': '#C62828', }, # ------------------------------------------------------------------ # MIX: Pair (1,2) is traffic-correlated; Pair (3,4) unrelated # beta_12 = 0.65^2 = 0.42; beta_i3, beta_i4 = 0 (diff scenes) # ------------------------------------------------------------------ 'MIX': { 'title': 'MIX Scenario (Correlated Pair + Unrelated Pair)', 'users': ['TL-Camera (U1)', 'Autovehicle (U2)', 'TV Viewer (U3)', 'Music Stream (U4)'], 'beta_u': [0.65, 0.65, 0.05, 0.05], 'scenes': ['traffic', 'traffic', 'home', 'office'], 'color': '#2E7D32', }, # ------------------------------------------------------------------ # HETERO: Three-tier heterogeneous correlation structure # U1-U2: beta_12 = 0.75^2 = 0.5625 (high, same HD camera) # U1-U3: beta_13 = 0.75*0.45 = 0.3375 (medium, same scene) # U1-U4: beta_14 = 0 (low, different context) # ------------------------------------------------------------------ 'HETERO': { 'title': 'HETERO Scenario (Heterogeneous Correlation Structure)', 'users': ['HD-Cam (U1)', 'HD-Cam (U2)', 'LR-Sensor (U3)', 'IoT (U4)'], 'beta_u': [0.75, 0.75, 0.45, 0.08], 'scenes': ['traffic', 'traffic', 'traffic', 'indoor'], 'color': '#6A1B9A', }, # ------------------------------------------------------------------ # ASYM: All users share one scene with a smooth beta gradient # beta_12=0.42, beta_13=0.25, beta_14=0.086, # beta_23=0.20, beta_24=0.070, beta_34=0.042 # ------------------------------------------------------------------ 'ASYM': { 'title': 'ASYM Scenario (Asymmetric Semantic Relevance)', 'users': ['U1 (beta=0.72)', 'U2 (beta=0.58)', 'U3 (beta=0.35)', 'U4 (beta=0.12)'], 'beta_u': [0.72, 0.58, 0.35, 0.12], 'scenes': ['traffic', 'traffic', 'traffic', 'traffic'], 'color': '#00695C', }, } # User color palette (consistent across figures) USER_COLORS = ['#1565C0', '#2E7D32', '#C62828', '#6A1B9A'] # Method display config: color / marker+linestyle / linewidth / legend label MCFG = { 'OFDMA': ('#546E7A', 's--', 1.5, 'OFDMA (Analytical)'), 'MAML+Attn': ('#1565C0', 'o-', 2.4, 'UWCA (analytical)'), } def compute_beta_matrix(cfg: dict) -> np.ndarray: """Compute the (U x U) semantic relevance matrix beta_uv = beta_u * beta_v for pairs sharing the same scene; zero otherwise.""" bu = np.array(cfg['beta_u']) sc = cfg['scenes'] buv = np.zeros((U, U)) for i in range(U): for j in range(U): if sc[i] == sc[j]: buv[i, j] = bu[i] * bu[j] return buv # ══════════════════════════════════════════════════════════════════════════════ # 2. Embedding generation (paper Eq. 2: x_u = sqrt(1-beta^2)*p_u + beta*s) # ══════════════════════════════════════════════════════════════════════════════ _SCENES: dict = {} # scene vector cache (reproducibility) def _get_scene(key: str) -> np.ndarray: if key not in _SCENES: v = RNG.standard_normal(D) _SCENES[key] = v / (np.linalg.norm(v) + 1e-8) return _SCENES[key] def gen_embeddings(n: int, scenario_key: str) -> np.ndarray: """Generate unit-normalized embeddings (n, U, D) for a named scenario. e_u = sqrt(1-beta_u^2) * p_hat_u + beta_u * s where p_hat_u is a unit-norm private vector (normalised before mixing), so ||e_u|| ≈ 1 and E[e_i[dim] · e_u[dim]] = beta_u·beta_i·||s[dim]||² (exact). """ cfg = SCENARIOS[scenario_key] bu = cfg['beta_u'] scenes = cfg['scenes'] embs = [] for u in range(U): s = _get_scene(scenes[u]) private = RNG.standard_normal((n, D)) p_hat = private / (np.linalg.norm(private, axis=-1, keepdims=True) + 1e-8) e = np.sqrt(1 - bu[u] ** 2) * p_hat + bu[u] * s[None, :] e /= np.linalg.norm(e, axis=-1, keepdims=True) + 1e-8 embs.append(e) return np.stack(embs, axis=1) # (n, U, D) def gen_embeddings_beta(n: int, beta: float) -> np.ndarray: """Generate embeddings where all users share a single scene at level beta. Used for the beta-sweep experiment (Proposition 1 validation).""" s = _get_scene('sweep') embs = [] for _ in range(U): private = RNG.standard_normal((n, D)) p_hat = private / (np.linalg.norm(private, axis=-1, keepdims=True) + 1e-8) e = np.sqrt(max(1 - beta ** 2, 0)) * p_hat + beta * s[None, :] e /= np.linalg.norm(e, axis=-1, keepdims=True) + 1e-8 embs.append(e) return np.stack(embs, axis=1) # ══════════════════════════════════════════════════════════════════════════════ # 3. Channel models # ══════════════════════════════════════════════════════════════════════════════ def rayleigh_channel(E: np.ndarray, snr_db: float) -> np.ndarray: """Rayleigh flat-fading channel: h ~ CN(0,1), AWGN noise.""" snr = 10 ** (snr_db / 10) h = (np.abs(RNG.standard_normal((*E.shape[:2], 1)) * np.sqrt(0.5) + 1j * RNG.standard_normal((*E.shape[:2], 1)) * np.sqrt(0.5)) ).real h = np.abs(h) noise_std = np.sqrt(np.mean(E ** 2) / snr) return h * E + RNG.standard_normal(E.shape) * noise_std def shared_embedding_channel(E: np.ndarray, snr_db: float) -> np.ndarray: """SE channel (JSAC shared-embedding framework). Masking: x_u = e_u ⊙ m_u Superpos.: y_tx = Σ_u x_u Reception: y_rx,u = h_u · y_tx + n_u (independent Rayleigh per user) Returns Y_rx of shape (n, U, D). """ n = E.shape[0] X = E * MASKS[None, :, :] # (n, U, D) masked Ytx = X.sum(axis=1) # (n, D) superimposed h = (np.sqrt(RNG.standard_normal((n, U, 1)) ** 2 + RNG.standard_normal((n, U, 1)) ** 2) * np.sqrt(0.5)) # Rayleigh |h|, (n,U,1) sig_power = float(np.mean(Ytx ** 2)) noise_std = np.sqrt(sig_power / (10 ** (snr_db / 10))) Yrx = h * Ytx[:, None, :] + RNG.standard_normal((n, U, D)) * noise_std return Yrx # (n, U, D) def noma_ul_channel(E: np.ndarray, snr_db: float): """NOMA uplink channel: all U users transmit to a single BS receiver. TX_u: x_u = sqrt(p_u) * e_u (power-scaled embedding) RX (BS): y = Σ_u h_u * x_u + n = Σ_u h_u * sqrt(p_u) * e_u + n (single D-dim received signal) Power allocation: NOMA_POWER = [0.40, 0.30, 0.20, 0.10] (descending, sum=1). Each user has an independent Rayleigh flat-fading channel h_u ~ Rayleigh(1/√2). Returns: y : (n, D) single received signal at BS h : (n, U, 1) per-user Rayleigh channel gains """ n = E.shape[0] h = (np.sqrt(RNG.standard_normal((n, U, 1)) ** 2 + RNG.standard_normal((n, U, 1)) ** 2) * np.sqrt(0.5)) # (n, U, 1) Rayleigh # Power-weighted, channel-scaled superposition at BS weighted = E * np.sqrt(NOMA_POWER)[None, :, None] * h # (n, U, D) y = weighted.sum(axis=1) # (n, D) BS received sig_power = float(np.mean(y ** 2)) noise_std = np.sqrt(sig_power / (10 ** (snr_db / 10))) y = y + RNG.standard_normal((n, D)) * noise_std return y, h def noma_sic_decoder(y: np.ndarray, h: np.ndarray) -> np.ndarray: """NOMA-SIC decoder at the BS for the uplink model. Decodes users in fixed descending allocated-power order (user 0 first, p=0.40; user 3 last, p=0.10). At each step u: 1. Equalize user u's channel in the current residual: z = residual / h_u 2. Decode: ê_u = normalize(z) 3. Subtract h_u * sqrt(p_u) * ê_u from the shared residual. Note: In the HIGH-correlation scenario (all β ≈ 0.65), SIC error propagation causes the weakest user (u=3) SER to *increase* at high SNR. This is a known fundamental limitation of NOMA-SIC under high semantic correlation: imperfect cancellation errors from stages 0–2 are fixed-magnitude (independent of SNR) and dominate the weakest user's residual once noise vanishes, creating an interference floor that worsens relative to the signal as SNR grows. Returns Eh: (n, U, D) decoded unit-norm embeddings. """ n = y.shape[0] Eh = np.zeros((n, U, D)) residual = y.copy() # (n, D) shared BS residual for u in range(U): # u=0: strongest, u=3: weakest z = residual / (h[:, u, :] + 1e-8) # (n, D) Eh[:, u, :] = _norm(z) residual -= h[:, u, :] * np.sqrt(NOMA_POWER[u]) * Eh[:, u, :] return Eh # ══════════════════════════════════════════════════════════════════════════════ # 4. Decoders (SE framework) # ══════════════════════════════════════════════════════════════════════════════ def _norm(E: np.ndarray) -> np.ndarray: return E / (np.linalg.norm(E, axis=-1, keepdims=True) + 1e-8) def ofdma_se_decoder(Y_rx: np.ndarray): """SE-OFDMA decoder: each user uses only their own D/U-dim subspace block. ê_u = normalize(y_rx,u ⊙ m_u) [extract own block only] Analytical cos_sim upper bound (high SNR, no noise): cos_sim(ê_u, e_u) = ||e_u ⊙ m_u|| ≈ sqrt(1/U) = 0.5 (U=4, any β) → Does NOT benefit from inter-user semantic correlation. → TAU must be set < 0.5 for SER to decrease with SNR. MI bound: I_OFDMA-SE = (D/U) · log2(1 + SNR_lin/D) """ Eh = np.stack([_norm(Y_rx[:, u, :] * MASKS[u]) for u in range(U)], axis=1) return Eh, None def maml_attention_se_decoder(Y_rx: np.ndarray, snr_db: float, beta_matrix: np.ndarray): """MAML cross-attention decoder (SE framework). Subspace extraction: R_{u,i} = y_rx,u ⊙ m_i Weights: α_{u,i} ∝ β_{u,i} (semantic relevance; self β_{u,u}=1) Output: ê_u = normalize(Σ_i α_{u,i}·R_{u,i} + y_rx,u ⊙ m_u) ↑ skip connection (extra self-emphasis) Key: weights do NOT collapse to diagonal at high SNR. Cross-user subspaces are always aggregated; their utility depends on β_{u,i}: HIGH scenario (β_uv ≈ 0.42): all subspaces carry scene info → full-D reconstruction. LOW scenario (β_uv ≈ 0.00): cross subspaces uninformative → gain ≈ 0. """ n, U_, D_ = Y_rx.shape # Subspace extractions: R[b, u, i, :] = Y_rx[b, u, :] * MASKS[i] R = Y_rx[:, :, None, :] * MASKS[None, None, :, :] # (n, U, U, D) # β-weighted attention (self = 1, cross = β_{u,i}) alpha = beta_matrix.copy() np.fill_diagonal(alpha, 1.0) alpha /= alpha.sum(1, keepdims=True) + 1e-8 # (U, U) row-normalised # Weighted aggregation: ctx[b, u, :] = Σ_i α_{u,i} · R[b, u, i, :] # Each block dims_i carries e_i[dims_i]; when β_ui is high, e_i[dims_i] ≈ e_u[dims_i] # → HIGH β: ctx ≈ e_u (full D dims reconstructed); LOW β: ctx ≈ own block only # NOTE: 'ui,buid->bud' — u (receiving user) and i (mask idx) summed over i only; # u is a free index kept in output so each user gets its own weighted sum. ctx = np.einsum('ui,buid->bud', alpha, R) # (n, U, D) Eh = np.stack([_norm(ctx[:, u, :]) for u in range(U_)], axis=1) return Eh, alpha.copy() # ── 4-B. S-step variant for ablation ───────────────────────────────────────── def maml_attention_se_decoder_S(Y_rx: np.ndarray, snr_db: float, beta_matrix: np.ndarray, S: int): """MAML-SE decoder parametrised by inner-loop steps S (ablation). S controls how well the decoder has learned β-selective weighting: S=0 → uniform weights across all U subspaces (no β awareness) S=5 → β-weighted (sweet-spot; matches maml_attention_se_decoder) S→∞ → same as S=5 (saturated) Interpolation: α = q·α_beta + (1-q)·α_uniform, q = 1-exp(-S/S_half) """ n, U_, D_ = Y_rx.shape S_HALF = 3.0 q = 1.0 - np.exp(-S / S_HALF) if S > 0 else 0.0 R = Y_rx[:, :, None, :] * MASKS[None, None, :, :] alpha_beta = beta_matrix.copy() np.fill_diagonal(alpha_beta, 1.0) alpha_beta /= alpha_beta.sum(1, keepdims=True) + 1e-8 alpha_uniform = np.ones((U_, U_)) / U_ alpha = q * alpha_beta + (1.0 - q) * alpha_uniform alpha /= alpha.sum(1, keepdims=True) + 1e-8 ctx = np.einsum('ui,buid->bud', alpha, R) Eh = np.stack([_norm(ctx[:, u, :]) for u in range(U_)], axis=1) return Eh, alpha # ══════════════════════════════════════════════════════════════════════════════ # 5. Metrics # ══════════════════════════════════════════════════════════════════════════════ def cos_sim(Eh: np.ndarray, Egt: np.ndarray) -> np.ndarray: return (Eh * Egt).sum(-1) # (n, U) def ser_total(Eh, Egt, tau=TAU) -> float: return float((cos_sim(Eh, Egt) < tau).mean()) def ser_per_user(Eh, Egt, tau=TAU) -> np.ndarray: return (cos_sim(Eh, Egt) < tau).mean(0) # (U,) def corr_matrix(Eh: np.ndarray) -> np.ndarray: """Mean pairwise cosine similarity matrix of decoded embeddings, shape (U, U). Since ê_u are unit-norm (from _norm), cos_sim(ê_u, ê_v) = ê_u · ê_v. Averaged over batch n. NOTE: Pearson correlation on mean vectors fails for SE framework because users operate in orthogonal subspaces. The mean-subtraction step creates a spurious negative offset in all inactive dims, making even orthogonal subspace vectors appear correlated. Cosine similarity is correct here. Expected values (high SNR): HIGH (all same scene, β=0.95): off-diag ≈ β² = 0.90 (all ê_u → s) LOW (different scenes): off-diag ≈ 0 (different scene directions, plus orthogonal subspace support) MIX (pair 1-2 correlated): off-diag[1,2] ≈ β², others ≈ 0 """ # Eh: (n, U, D), already unit-norm from _norm return np.einsum('nud,nvd->uv', Eh, Eh) / Eh.shape[0] # ══════════════════════════════════════════════════════════════════════════════ # 6. Simulation loops # ══════════════════════════════════════════════════════════════════════════════ def run_scenario(scenario_key: str) -> dict: """Run full SNR sweep for one scenario; returns SER/per-user/rho/attn dicts.""" cfg = SCENARIOS[scenario_key] beta_mat = compute_beta_matrix(cfg) methods = ['OFDMA', 'NOMA-SIC', 'MAML+Attn'] res = {m: {'ser': [], 'sp': []} for m in methods} rho_m = [] attn_m_sum = np.zeros((U, U)) cnt10 = 0 print(f" [{scenario_key:6s}]", end='', flush=True) for si, snr in enumerate(SNR_DB): acc = {m: {'ser': 0., 'sp': np.zeros(U)} for m in methods} for _ in range(N_MC): Egt = gen_embeddings(BATCH, scenario_key) # -- OFDMA-SE: own D/U-dim block, same SE channel, no SNR penalty --- Yrx_ofdma = shared_embedding_channel(Egt, snr) Eh, _ = ofdma_se_decoder(Yrx_ofdma) acc['OFDMA']['ser'] += ser_total(Eh, Egt) acc['OFDMA']['sp'] += ser_per_user(Eh, Egt) # -- NOMA-SIC: uplink, power-weighted TX, SIC at BS ------------------ y_noma, h_noma = noma_ul_channel(Egt, snr) Eh_noma = noma_sic_decoder(y_noma, h_noma) acc['NOMA-SIC']['ser'] += ser_total(Eh_noma, Egt) acc['NOMA-SIC']['sp'] += ser_per_user(Eh_noma, Egt) # -- MAML+Attn-SE: cross-attention over all subspaces -------------- Yrx = shared_embedding_channel(Egt, snr) Eh, am = maml_attention_se_decoder(Yrx, float(snr), beta_mat) acc['MAML+Attn']['ser'] += ser_total(Eh, Egt) acc['MAML+Attn']['sp'] += ser_per_user(Eh, Egt) if si == 5: # SNR = 10 dB index rho_m.append(corr_matrix(Eh)) attn_m_sum += am; cnt10 += 1 for m in methods: res[m]['ser'].append(acc[m]['ser'] / N_MC) res[m]['sp'].append(acc[m]['sp'] / N_MC) if (si + 1) % 3 == 0: print('.', end='', flush=True) for m in methods: res[m]['ser'] = np.array(res[m]['ser']) res[m]['sp'] = np.array(res[m]['sp']) n10 = max(cnt10, 1) res['_rho_m'] = np.mean(rho_m, axis=0) if rho_m else np.eye(U) res['_attn_m'] = attn_m_sum / n10 res['_beta_mat'] = beta_mat return res def run_beta_sweep(beta_values: np.ndarray, snr_db: float = 10.0) -> dict: """SER gain vs beta sweep (validates Proposition 1: monotone gain).""" gain_maml = [] for beta in beta_values: s_ofdma = s_maml = 0. beta_mat = beta ** 2 * np.ones((U, U)) np.fill_diagonal(beta_mat, 1.0) for _ in range(N_MC): Egt = gen_embeddings_beta(BATCH, beta) Yrx_o = shared_embedding_channel(Egt, snr_db) Eh, _ = ofdma_se_decoder(Yrx_o) s_ofdma += ser_total(Eh, Egt) Yrx = shared_embedding_channel(Egt, snr_db) Eh, _ = maml_attention_se_decoder(Yrx, snr_db, beta_mat) s_maml += ser_total(Eh, Egt) gain_maml.append((s_ofdma - s_maml) / N_MC) return {'gain_maml': np.array(gain_maml)} def run_ablation(S_values: list, snr_db: float = 10.0, scenario_key: str = 'HIGH') -> dict: """MAML inner-loop steps S ablation study at a fixed SNR point.""" cfg = SCENARIOS[scenario_key] beta_mat = compute_beta_matrix(cfg) ser_list = [] print(f" [ablation S-sweep]", end='', flush=True) for S in S_values: s_acc = 0. for _ in range(N_MC): Egt = gen_embeddings(BATCH, scenario_key) Yrx = shared_embedding_channel(Egt, snr_db) Eh, _ = maml_attention_se_decoder_S(Yrx, snr_db, beta_mat, S) s_acc += ser_total(Eh, Egt) ser_list.append(s_acc / N_MC) print('.', end='', flush=True) # Ideal MAML baseline (S -> inf) s_ideal = 0. for _ in range(N_MC): Egt = gen_embeddings(BATCH, scenario_key) Yrx = shared_embedding_channel(Egt, snr_db) Eh, _ = maml_attention_se_decoder(Yrx, snr_db, beta_mat) s_ideal += ser_total(Eh, Egt) return {'S_values': S_values, 'ser': np.array(ser_list), 'ser_ideal': s_ideal / N_MC} # ══════════════════════════════════════════════════════════════════════════════ # 6-B. Mutual Information analysis (analytical, Proposition 2) # ══════════════════════════════════════════════════════════════════════════════ def _erg_cap(a_arr: np.ndarray) -> np.ndarray: """Ergodic capacity E[log2(1 + a·h²)] bits, h² ~ Exp(1) (Rayleigh, E[h²]=1). Closed form: C_erg(a) = exp(1/a) · E1(1/a) / ln(2) [a > 0] Derivation: ∫₀^∞ log₂(1+a·x)·e^{-x}dx = e^{1/a}·E1(1/a)/ln(2) Limits: a → 0 : C_erg ≈ a/ln(2) (linear in SNR) a → ∞ : C_erg ≈ log₂(a) − γ_E/ln(2) (γ_E ≈ 0.5772, logarithmic) """ a = np.asarray(a_arr, dtype=float) inv_a = np.where(a > 1e-30, 1.0 / a, 1e30) return np.exp(inv_a) * exp1(inv_a) / np.log(2) def mutual_information_bounds(snr_db_arr: np.ndarray, beta: float, U_val: int = 4, D_val: int = 64) -> dict: """Ergodic MI bounds under Rayleigh fading (bits per channel use per user). Channel: y_rx,u = h_u · y_tx + n, h_u ~ CN(0,1) → |h_u|² ~ Exp(1) Signal power: E[||y_tx||²] = 1, noise σ² = 1/SNR_lin per element. ── OFDMA-SE (own DPU-dim block only) ──────────────────────────────── I_OFDMA = DPU · E[log₂(1 + |h|²·SNR/D)] = DPU · C_erg(SNR/D) [ergodic Rayleigh] ── UWCA-SE (all D dims via β-weighted cross-attention) ────────────── Own block (i = u): DPU dims, same as OFDMA Cross block (i ≠ u): DPU dims, effective ergodic SINR capacity: I_cross = DPU · E[log₂(1 + β²·|h|²·SNR / (D + (1-β²)·|h|²·SNR))] Using E[log₂(1 + β²·x/(D/SNR + (1-β²)·x))] = E[log₂(1 + x·SNR/D)] − E[log₂(1 + (1-β²)·x·SNR/D)] = C_erg(SNR/D) − C_erg((1-β²)·SNR/D) [subtraction form] I_UWCA = DPU · C_erg(SNR/D) + (U-1)·DPU · [C_erg(SNR/D) − C_erg((1-β²)·SNR/D)] ── MI ratio properties ────────────────────────────────────────────── Low-SNR (SNR→0): ratio → 1 + (U-1)·β² ← maximum High-SNR (SNR→∞): ratio → 1 ← cross-block SINR saturates [because C_erg(SNR/D)−C_erg((1-β²)·SNR/D) → log₂(1/(1-β²)) = const] The ratio is strictly DECREASING in SNR; it is bounded in [1, 1+(U-1)·β²]. NOTE: "U·β²" is NOT the correct limit at any SNR regime. """ snr_lin = 10 ** (snr_db_arr / 10) DPU = D_val // U_val a_own = snr_lin / D_val # own-block SNR per dim a_priv = (1 - beta**2) * snr_lin / D_val # private-only SNR per dim C_own = _erg_cap(a_own) # E[log₂(1+|h|²·a_own)] C_priv = _erg_cap(a_priv) # E[log₂(1+|h|²·a_priv)] I_ofdma = DPU * C_own I_cross = DPU * (C_own - C_priv) # ergodic cross-block gain I_uwca = I_ofdma + (U_val - 1) * I_cross # Low-SNR analytical limit for the ratio (monotone decreasing in SNR) ratio_low_snr = 1.0 + (U_val - 1) * beta**2 # SNR→0 limit return {'I_ofdma': I_ofdma, 'I_uwca': I_uwca, 'snr_db': snr_db_arr, 'U': U_val, 'beta': beta, 'ratio_low_snr': ratio_low_snr} # ══════════════════════════════════════════════════════════════════════════════ # 6-C. U-user scaling experiment (U = 1, 2, 3, 4) # ══════════════════════════════════════════════════════════════════════════════ def run_u_variation(beta: float = 0.95, snr_db_arr: np.ndarray = None, n_mc: int = None, batch: int = None) -> dict: """SER vs SNR for U = 1, 2, 3, 4 users (HIGH correlation, all same scene). Analytical cos_sim bounds (high SNR): OFDMA-SE: cos_sim → sqrt(1/U) {U=1: 1.00, U=2: 0.71, U=4: 0.50} UWCA-SE: cos_sim → sqrt(1/U + (U-1)β²/U) = sqrt((1+(U-1)β²)/U) {U=1: 1.00, U=2: 0.95, U=4: 0.96} MI ratio (high SNR): I_UWCA / I_OFDMA → U · β² (scales linearly with U) """ if snr_db_arr is None: snr_db_arr = SNR_DB n_mc = n_mc if n_mc is not None else N_MC batch = batch if batch is not None else BATCH U_list = [1, 2, 3, 4] results_u = {} print(" [U-variation]", end='', flush=True) for U_val in U_list: DPU_val = D // U_val # Local orthogonal masks for this U masks_loc = np.zeros((U_val, D)) for _u in range(U_val): masks_loc[_u, _u * DPU_val : (_u + 1) * DPU_val] = 1.0 # β-matrix: all users same scene (HIGH) beta_mat = beta ** 2 * np.ones((U_val, U_val)) np.fill_diagonal(beta_mat, 1.0) res = {'OFDMA': {'ser': []}, 'UWCA': {'ser': []}} scene_vec = _get_scene(f'traffic_uvar_{U_val}') for snr in snr_db_arr: acc = {'OFDMA': 0., 'UWCA': 0.} for _ in range(n_mc): # Generate HIGH-correlated embeddings for U_val users E_list = [] for u in range(U_val): priv = RNG.standard_normal((batch, D)) p_h = priv / (np.linalg.norm(priv, axis=-1, keepdims=True) + 1e-8) e = np.sqrt(1 - beta**2) * p_h + beta * scene_vec[None, :] e /= np.linalg.norm(e, axis=-1, keepdims=True) + 1e-8 E_list.append(e) Egt = np.stack(E_list, axis=1) # (batch, U_val, D) # SE channel (U_val users) X = Egt * masks_loc[None, :, :] # (batch, U_val, D) masked Ytx = X.sum(axis=1) # (batch, D) superimposed n_b = batch h = (np.sqrt(RNG.standard_normal((n_b, U_val, 1)) ** 2 + RNG.standard_normal((n_b, U_val, 1)) ** 2) * np.sqrt(0.5)) sp = float(np.mean(Ytx ** 2)) nstd = np.sqrt(sp / (10 ** (snr / 10))) Yrx = h * Ytx[:, None, :] + RNG.standard_normal((n_b, U_val, D)) * nstd # OFDMA-SE decoder: own block only Eh_o = np.stack([_norm(Yrx[:, u, :] * masks_loc[u]) for u in range(U_val)], axis=1) acc['OFDMA'] += ser_total(Eh_o, Egt) # UWCA-SE decoder: β-weighted cross-attention over all U_val blocks R = Yrx[:, :, None, :] * masks_loc[None, None, :, :] # (B,U,U,D) alpha = beta_mat.copy() np.fill_diagonal(alpha, 1.0) alpha /= alpha.sum(1, keepdims=True) + 1e-8 ctx = np.einsum('ni,bnid->bnd', alpha, R) # (B,U,D) Eh_u = np.stack([_norm(ctx[:, u, :]) for u in range(U_val)], axis=1) acc['UWCA'] += ser_total(Eh_u, Egt) res['OFDMA']['ser'].append(acc['OFDMA'] / n_mc) res['UWCA']['ser'].append(acc['UWCA'] / n_mc) if snr == snr_db_arr[-1]: print('.', end='', flush=True) res['OFDMA']['ser'] = np.array(res['OFDMA']['ser']) res['UWCA']['ser'] = np.array(res['UWCA']['ser']) results_u[U_val] = res print() return results_u # ══════════════════════════════════════════════════════════════════════════════ # 7. Run all experiments # ══════════════════════════════════════════════════════════════════════════════ print("=" * 60) print("Semantic Correlation Simulation") print(f" d={D}, U={U}, N_MC={N_MC}, BATCH={BATCH}") print("=" * 60) results = {} for sk in SCENARIOS: results[sk] = run_scenario(sk) print() # newline after dots print(" [beta sweep]", end='', flush=True) BETA_VALUES = np.linspace(0.0, 0.9, 19) SWEEP_SNRS = [0.0, 5.0, 10.0] beta_sweeps = {snr: run_beta_sweep(BETA_VALUES, snr_db=snr) for snr in SWEEP_SNRS} beta_sweep = beta_sweeps[10.0] # backward-compat alias print(" done") S_VALUES = [1, 2, 3, 5, 7, 10, 15] ablation = run_ablation(S_VALUES, snr_db=10.0, scenario_key='MIX') print(" done") # Exp 6: U-variation (HIGH scenario, for fig9/fig10) u_var_results = run_u_variation(beta=0.95) # Exp 6-fig12: high-precision U-variation for fig12 (β=0.9/0.5/0.1, 1-dB SNR grid) _SNR_F12 = np.arange(0, 21, 1) # 1-dB step → smoother curves _N_MC_F12 = 1500 # 1500 × 256 = 384,000 samples/SNR point _BATCH_F12 = 256 print(" [fig12 high-precision U-variation β=0.9]", end='', flush=True) u_var_f12_09 = run_u_variation(beta=0.9, snr_db_arr=_SNR_F12, n_mc=_N_MC_F12, batch=_BATCH_F12) print(" [fig12 high-precision U-variation β=0.5]", end='', flush=True) u_var_f12_05 = run_u_variation(beta=0.5, snr_db_arr=_SNR_F12, n_mc=_N_MC_F12, batch=_BATCH_F12) print(" [fig12 high-precision U-variation β=0.1]", end='', flush=True) u_var_f12_01 = run_u_variation(beta=0.1, snr_db_arr=_SNR_F12, n_mc=_N_MC_F12, batch=_BATCH_F12) # Exp 6b: U-variation for LOW scenario # beta_uv ≈ 0 (independent scenes) → UWCA attention → diagonal → OFDMA-like def run_u_variation_low(snr_db_arr=None): """U-variation for LOW scenario: each user has an independent scene, beta_u ≈ 0. Expected: UWCA-SE ≈ OFDMA (attention collapses to identity mask) Statistical gain still present (U decreases → more dims per user → SER drops) """ if snr_db_arr is None: snr_db_arr = SNR_DB BETA_LOW = 0.05 # near-zero semantic relevance U_list = [1, 2, 3, 4] results_low = {} print(" [U-variation LOW]", end='', flush=True) for U_val in U_list: DPU_val = D // U_val masks_loc = np.zeros((U_val, D)) for _u in range(U_val): masks_loc[_u, _u * DPU_val:(_u + 1) * DPU_val] = 1.0 # beta matrix: near-zero off-diagonal → attention ≈ identity beta_mat_low = BETA_LOW ** 2 * np.ones((U_val, U_val)) np.fill_diagonal(beta_mat_low, 1.0) alpha_low = beta_mat_low / beta_mat_low.sum(axis=1, keepdims=True) res = {'OFDMA': {'ser': []}, 'UWCA': {'ser': []}} for snr in snr_db_arr: acc = {'OFDMA': 0., 'UWCA': 0.} for _ in range(N_MC): # Each user has its OWN independent scene (LOW scenario) E_list = [] for u in range(U_val): scene_u = _get_scene(f'low_uvar_{U_val}_{u}') priv = RNG.standard_normal((BATCH, D)) p_h = priv / (np.linalg.norm(priv, axis=-1, keepdims=True) + 1e-8) e = np.sqrt(1 - BETA_LOW ** 2) * p_h + BETA_LOW * scene_u[None, :] e /= np.linalg.norm(e, axis=-1, keepdims=True) + 1e-8 E_list.append(e) Egt = np.stack(E_list, axis=1) # (BATCH, U_val, D) X = Egt * masks_loc[None, :, :] Ytx = X.sum(axis=1) h = (np.sqrt(RNG.standard_normal((BATCH, U_val, 1)) ** 2 + RNG.standard_normal((BATCH, U_val, 1)) ** 2) * np.sqrt(0.5)) sp = float(np.mean(Ytx ** 2)) nstd = np.sqrt(sp / (10 ** (snr / 10))) Yrx = h * Ytx[:, None, :] + RNG.standard_normal((BATCH, U_val, D)) * nstd # OFDMA-SE ehat_o = np.stack([_norm(Yrx[:, u, :] * masks_loc[u]) for u in range(U_val)], axis=1) cos_o = np.sum(ehat_o * Egt, axis=-1) acc['OFDMA'] += np.mean(cos_o < TAU) # UWCA-SE (near-diagonal attention → OFDMA-like) R = Yrx[:, :, None, :] * masks_loc[None, None, :, :] ctx = np.einsum('ui,buid->bud', alpha_low, R) ehat_w = np.stack([_norm(ctx[:, u, :]) for u in range(U_val)], axis=1) cos_w = np.sum(ehat_w * Egt, axis=-1) acc['UWCA'] += np.mean(cos_w < TAU) res['OFDMA']['ser'].append(acc['OFDMA'] / N_MC) res['UWCA']['ser'].append(acc['UWCA'] / N_MC) results_low[U_val] = {k: {'ser': np.array(v['ser'])} for k, v in res.items()} print('.', end='', flush=True) print(' done') return results_low u_var_low_results = run_u_variation_low() # Exp 7: MI bounds for U = 1, 2, 3, 4 (analytical) MI_SNRS = np.linspace(0, 20, 200) MI_U_LIST = [1, 2, 3, 4] mi_bounds = {U_v: mutual_information_bounds(MI_SNRS, beta=0.95, U_val=U_v) for U_v in MI_U_LIST} print(f" [MI bounds computed for U={MI_U_LIST}]") print() # Frequently used indices IDX10 = int(np.argmin(np.abs(SNR_DB - 10))) IDX4 = int(np.argmin(np.abs(SNR_DB - 4))) IDX16 = int(np.argmin(np.abs(SNR_DB - 16))) mask = ~np.eye(U, dtype=bool) BETAS2 = BETA_VALUES ** 2 # Fair comparison experiment (needed before saving) D_SRC_F = D // U # = 16 per-user source embedding dimension (fixed) D_CH_F = D # = 64 total channel dimension (fixed) TAU_FAIR = 0.85 # SER threshold for fair comparison BETA_FAIR = 0.95 # HIGH semantic correlation scenario _U_LIST_F = [1, 2, 4] # Fixed reference power: power one user contributes per channel dim (independent of U) _REF_PWR_F = D_SRC_F / D_CH_F # = 0.25 def run_fair_u(U_val): """Fair comparison simulation for U_val users. Source: e_u ∈ ℝ^{D_SRC_F=16}, unit-norm. TX: block-placed into ℝ^{D_CH_F=64}; no actual interference (orthogonal blocks). RX: Rayleigh per-user, fixed noise_std independent of U. OFDMA: extract own 16-dim block, cos_sim in ℝ^16 → no structural ceiling. UWCA: aggregate all U 16-dim blocks via β-weighted cross-attn, cos_sim in ℝ^16. """ if U_val == 1: alpha_f = np.ones((1, 1)) else: bm = np.full((U_val, U_val), BETA_FAIR ** 2) np.fill_diagonal(bm, 1.0) alpha_f = bm / bm.sum(axis=1, keepdims=True) ser_o, ser_w = [], [] for snr in SNR_DB: snr_lin = 10 ** (snr / 10) noise_std = np.sqrt(_REF_PWR_F / snr_lin) # fixed, independent of U cos_o_all, cos_w_all = [], [] for _ in range(N_MC): # --- Source embeddings (BATCH, U_val, D_SRC_F) --- s = RNG.standard_normal(D_SRC_F) s /= np.linalg.norm(s) + 1e-8 E = np.zeros((BATCH, U_val, D_SRC_F)) for u in range(U_val): p = RNG.standard_normal((BATCH, D_SRC_F)) p /= np.linalg.norm(p, axis=-1, keepdims=True) + 1e-8 e = np.sqrt(1 - BETA_FAIR ** 2) * p + BETA_FAIR * s[None, :] E[:, u, :] = e / (np.linalg.norm(e, axis=-1, keepdims=True) + 1e-8) # --- Block placement into D_CH_F-dim channel --- Y_tx = np.zeros((BATCH, D_CH_F)) for u in range(U_val): Y_tx[:, u * D_SRC_F:(u + 1) * D_SRC_F] = E[:, u, :] # --- Rayleigh per-user fading --- h = (np.sqrt(RNG.standard_normal((BATCH, U_val, 1)) ** 2 + RNG.standard_normal((BATCH, U_val, 1)) ** 2) * np.sqrt(0.5)) Y_rx = (h * Y_tx[:, None, :] + RNG.standard_normal((BATCH, U_val, D_CH_F)) * noise_std) # Y_rx: (BATCH, U_val, D_CH_F) # --- OFDMA (fair): own 16-dim block only, cos_sim in ℝ^16 --- for u in range(U_val): blk = Y_rx[:, u, u * D_SRC_F:(u + 1) * D_SRC_F] # (BATCH, 16) ehat = blk / (np.linalg.norm(blk, axis=-1, keepdims=True) + 1e-8) cos_o_all.append((ehat * E[:, u, :]).sum(-1)) # --- UWCA-SE (fair): aggregate all U 16-dim blocks, cos_sim in ℝ^16 --- for u in range(U_val): ctx = np.zeros((BATCH, D_SRC_F)) for i in range(U_val): blk_i = Y_rx[:, u, i * D_SRC_F:(i + 1) * D_SRC_F] ctx += alpha_f[u, i] * blk_i ehat = ctx / (np.linalg.norm(ctx, axis=-1, keepdims=True) + 1e-8) cos_w_all.append((ehat * E[:, u, :]).sum(-1)) ser_o.append(float(np.mean(np.concatenate(cos_o_all) < TAU_FAIR))) ser_w.append(float(np.mean(np.concatenate(cos_w_all) < TAU_FAIR))) return np.array(ser_o), np.array(ser_w) print(" [Fair comparison (fig11)]", end='', flush=True) fair_results = {} for _U in _U_LIST_F: _so, _sw = run_fair_u(_U) fair_results[_U] = {'OFDMA': _so, 'UWCA': _sw} print('.', end='', flush=True) print(' done') # ══════════════════════════════════════════════════════════════════════════════ # 8. Save all results to CSV # ══════════════════════════════════════════════════════════════════════════════ try: import pandas as pd _USE_PANDAS = True except ImportError: _USE_PANDAS = False DATA_DIR = f'{OUT_DIR}/data' def _save_csv(df_or_dict, filename, columns=None): """Save a DataFrame (or dict of arrays) to CSV.""" path = os.path.join(DATA_DIR, filename) if _USE_PANDAS: if isinstance(df_or_dict, dict): df = pd.DataFrame(df_or_dict, columns=columns) else: df = df_or_dict df.to_csv(path, index=False) else: # Fallback: numpy if isinstance(df_or_dict, dict): arr = np.column_stack([df_or_dict[c] for c in columns]) header = ','.join(columns) else: arr = df_or_dict header = ','.join(columns) if columns else '' np.savetxt(path, arr, delimiter=',', header=header, comments='') print(f" Saved: {path}") # --- snr_db.csv --- _save_csv({'snr_db': SNR_DB}, 'snr_db.csv', columns=['snr_db']) # --- snr_f12.csv --- _save_csv({'snr_db': _SNR_F12}, 'snr_f12.csv', columns=['snr_db']) # --- mi_snrs.csv --- _save_csv({'snr_db': MI_SNRS}, 'mi_snrs.csv', columns=['snr_db']) # --- ser_scenarios.csv --- # columns: snr_db, scenario, method, ser _rows_ser = [] for sk in SCENARIOS: for m in ['OFDMA', 'NOMA-SIC', 'MAML+Attn']: for si, snr in enumerate(SNR_DB): _rows_ser.append({ 'snr_db': float(snr), 'scenario': sk, 'method': m, 'ser': float(results[sk][m]['ser'][si]), }) if _USE_PANDAS: pd.DataFrame(_rows_ser).to_csv(os.path.join(DATA_DIR, 'ser_scenarios.csv'), index=False) print(f" Saved: {DATA_DIR}/ser_scenarios.csv") else: _cols = ['snr_db', 'scenario', 'method', 'ser'] with open(os.path.join(DATA_DIR, 'ser_scenarios.csv'), 'w') as _f: _f.write(','.join(_cols) + '\n') for r in _rows_ser: _f.write(f"{r['snr_db']},{r['scenario']},{r['method']},{r['ser']}\n") print(f" Saved: {DATA_DIR}/ser_scenarios.csv") # --- attn_heatmaps.csv --- # columns: scenario, row, col, alpha _rows_attn = [] for sk in ['HIGH', 'LOW', 'MIX']: am = results[sk]['_attn_m'] for i in range(U): for j in range(U): _rows_attn.append({ 'scenario': sk, 'row': i, 'col': j, 'alpha': float(am[i, j]), }) if _USE_PANDAS: pd.DataFrame(_rows_attn).to_csv(os.path.join(DATA_DIR, 'attn_heatmaps.csv'), index=False) print(f" Saved: {DATA_DIR}/attn_heatmaps.csv") else: _cols = ['scenario', 'row', 'col', 'alpha'] with open(os.path.join(DATA_DIR, 'attn_heatmaps.csv'), 'w') as _f: _f.write(','.join(_cols) + '\n') for r in _rows_attn: _f.write(f"{r['scenario']},{r['row']},{r['col']},{r['alpha']}\n") print(f" Saved: {DATA_DIR}/attn_heatmaps.csv") # --- ser_per_user_mix.csv --- # columns: snr_db, user, method, ser _rows_puser = [] res_mix = results['MIX'] for si, snr in enumerate(SNR_DB): for ui in range(U): for m in ['OFDMA', 'NOMA-SIC', 'MAML+Attn']: _rows_puser.append({ 'snr_db': float(snr), 'user': ui, 'method': m, 'ser': float(res_mix[m]['sp'][si, ui]), }) if _USE_PANDAS: pd.DataFrame(_rows_puser).to_csv(os.path.join(DATA_DIR, 'ser_per_user_mix.csv'), index=False) print(f" Saved: {DATA_DIR}/ser_per_user_mix.csv") else: _cols = ['snr_db', 'user', 'method', 'ser'] with open(os.path.join(DATA_DIR, 'ser_per_user_mix.csv'), 'w') as _f: _f.write(','.join(_cols) + '\n') for r in _rows_puser: _f.write(f"{r['snr_db']},{r['user']},{r['method']},{r['ser']}\n") print(f" Saved: {DATA_DIR}/ser_per_user_mix.csv") # --- beta_sweep.csv --- # columns: beta_sq, snr_label, ser_ofdma, ser_joint, ser_maml, gain_ofdma, gain_joint, gain_maml # ser_ofdma/joint are not computed in this sim (only gain_maml); fill zeros for compat _rows_beta = [] for bi, bsq in enumerate(BETAS2): for snr_lbl in SWEEP_SNRS: gm = float(beta_sweeps[snr_lbl]['gain_maml'][bi]) _rows_beta.append({ 'beta_sq': float(bsq), 'snr_label': float(snr_lbl), 'ser_ofdma': 0.0, 'ser_joint': 0.0, 'ser_maml': 0.0, 'gain_ofdma': 0.0, 'gain_joint': 0.0, 'gain_maml': gm, }) if _USE_PANDAS: pd.DataFrame(_rows_beta).to_csv(os.path.join(DATA_DIR, 'beta_sweep.csv'), index=False) print(f" Saved: {DATA_DIR}/beta_sweep.csv") else: _cols = ['beta_sq', 'snr_label', 'ser_ofdma', 'ser_joint', 'ser_maml', 'gain_ofdma', 'gain_joint', 'gain_maml'] with open(os.path.join(DATA_DIR, 'beta_sweep.csv'), 'w') as _f: _f.write(','.join(_cols) + '\n') for r in _rows_beta: _f.write(','.join(str(r[c]) for c in _cols) + '\n') print(f" Saved: {DATA_DIR}/beta_sweep.csv") # --- ablation.csv --- # columns: S, ser # S=999 reserved for ser_ideal _rows_abl = [{'S': int(s), 'ser': float(sv)} for s, sv in zip(ablation['S_values'], ablation['ser'])] _rows_abl.append({'S': 999, 'ser': float(ablation['ser_ideal'])}) if _USE_PANDAS: pd.DataFrame(_rows_abl).to_csv(os.path.join(DATA_DIR, 'ablation.csv'), index=False) print(f" Saved: {DATA_DIR}/ablation.csv") else: with open(os.path.join(DATA_DIR, 'ablation.csv'), 'w') as _f: _f.write('S,ser\n') for r in _rows_abl: _f.write(f"{r['S']},{r['ser']}\n") print(f" Saved: {DATA_DIR}/ablation.csv") # --- u_variation_high.csv --- # columns: snr_db, U, method, ser _rows_uvar_high = [] for U_val in [1, 2, 3, 4]: for si, snr in enumerate(SNR_DB): for m in ['OFDMA', 'UWCA']: _rows_uvar_high.append({ 'snr_db': float(snr), 'U': U_val, 'method': m, 'ser': float(u_var_results[U_val][m]['ser'][si]), }) if _USE_PANDAS: pd.DataFrame(_rows_uvar_high).to_csv(os.path.join(DATA_DIR, 'u_variation_high.csv'), index=False) print(f" Saved: {DATA_DIR}/u_variation_high.csv") else: with open(os.path.join(DATA_DIR, 'u_variation_high.csv'), 'w') as _f: _f.write('snr_db,U,method,ser\n') for r in _rows_uvar_high: _f.write(f"{r['snr_db']},{r['U']},{r['method']},{r['ser']}\n") print(f" Saved: {DATA_DIR}/u_variation_high.csv") # --- u_variation_f12.csv --- # columns: snr_db, beta, U, method, ser _rows_uvar_f12 = [] for _beta_val, _uvar_dict in [(0.9, u_var_f12_09), (0.5, u_var_f12_05), (0.1, u_var_f12_01)]: for U_val in [1, 2, 3, 4]: for si, snr in enumerate(_SNR_F12): for m in ['OFDMA', 'UWCA']: _rows_uvar_f12.append({ 'snr_db': float(snr), 'beta': _beta_val, 'U': U_val, 'method': m, 'ser': float(_uvar_dict[U_val][m]['ser'][si]), }) if _USE_PANDAS: pd.DataFrame(_rows_uvar_f12).to_csv(os.path.join(DATA_DIR, 'u_variation_f12.csv'), index=False) print(f" Saved: {DATA_DIR}/u_variation_f12.csv") else: with open(os.path.join(DATA_DIR, 'u_variation_f12.csv'), 'w') as _f: _f.write('snr_db,beta,U,method,ser\n') for r in _rows_uvar_f12: _f.write(f"{r['snr_db']},{r['beta']},{r['U']},{r['method']},{r['ser']}\n") print(f" Saved: {DATA_DIR}/u_variation_f12.csv") # --- u_variation_low.csv --- # columns: snr_db, U, method, ser _rows_uvar_low = [] for U_val in [1, 2, 3, 4]: for si, snr in enumerate(SNR_DB): for m in ['OFDMA', 'UWCA']: _rows_uvar_low.append({ 'snr_db': float(snr), 'U': U_val, 'method': m, 'ser': float(u_var_low_results[U_val][m]['ser'][si]), }) if _USE_PANDAS: pd.DataFrame(_rows_uvar_low).to_csv(os.path.join(DATA_DIR, 'u_variation_low.csv'), index=False) print(f" Saved: {DATA_DIR}/u_variation_low.csv") else: with open(os.path.join(DATA_DIR, 'u_variation_low.csv'), 'w') as _f: _f.write('snr_db,U,method,ser\n') for r in _rows_uvar_low: _f.write(f"{r['snr_db']},{r['U']},{r['method']},{r['ser']}\n") print(f" Saved: {DATA_DIR}/u_variation_low.csv") # --- mi_bounds.csv --- # columns: snr_db, U, I_ofdma, I_uwca, ratio_low_snr _rows_mi = [] for U_val in MI_U_LIST: mb = mi_bounds[U_val] rl = float(mb['ratio_low_snr']) for si, snr in enumerate(MI_SNRS): _rows_mi.append({ 'snr_db': float(snr), 'U': U_val, 'I_ofdma': float(mb['I_ofdma'][si]), 'I_uwca': float(mb['I_uwca'][si]), 'ratio_low_snr': rl, }) if _USE_PANDAS: pd.DataFrame(_rows_mi).to_csv(os.path.join(DATA_DIR, 'mi_bounds.csv'), index=False) print(f" Saved: {DATA_DIR}/mi_bounds.csv") else: with open(os.path.join(DATA_DIR, 'mi_bounds.csv'), 'w') as _f: _f.write('snr_db,U,I_ofdma,I_uwca,ratio_low_snr\n') for r in _rows_mi: _f.write(f"{r['snr_db']},{r['U']},{r['I_ofdma']},{r['I_uwca']},{r['ratio_low_snr']}\n") print(f" Saved: {DATA_DIR}/mi_bounds.csv") # --- fair_comparison.csv --- # columns: snr_db, U, method, ser _rows_fair = [] for U_val in _U_LIST_F: for si, snr in enumerate(SNR_DB): for m in ['OFDMA', 'UWCA']: _rows_fair.append({ 'snr_db': float(snr), 'U': U_val, 'method': m, 'ser': float(fair_results[U_val][m.upper()][si]), }) if _USE_PANDAS: pd.DataFrame(_rows_fair).to_csv(os.path.join(DATA_DIR, 'fair_comparison.csv'), index=False) print(f" Saved: {DATA_DIR}/fair_comparison.csv") else: with open(os.path.join(DATA_DIR, 'fair_comparison.csv'), 'w') as _f: _f.write('snr_db,U,method,ser\n') for r in _rows_fair: _f.write(f"{r['snr_db']},{r['U']},{r['method']},{r['ser']}\n") print(f" Saved: {DATA_DIR}/fair_comparison.csv") print() print("All data saved to results/data/")