1221 lines
56 KiB
Python
Executable File
1221 lines
56 KiB
Python
Executable File
"""
|
||
=============================================================================
|
||
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/")
|