hc/SemanticCommunication
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: add project materials and code

Browse Source
This commit is contained in:
hc
2026-02-28 16:17:42 +08:00
commit 5efb877df7
198 changed files with 17541 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
code/envs/__init__.py
+6
View File
@@ -0,0 +1,6 @@
"""Environment modules for Co-MADDPG wireless resource allocation."""
from .channel_model import ChannelModel
from .semantic_module import SemanticModule
__all__ = ["ChannelModel", "SemanticModule"]
code/envs/channel_model.py
+197
View File
@@ -0,0 +1,197 @@
"""
无线资源分配信道模型 / Channel model for OFDMA wireless resource allocation.
该模块实现了 3GPP 风格的路径损耗模型和多用户 OFDMA 下行链路系统的复信道增益生成。
所有公式遵循论文中的公式 (5)-(8)。
This module implements the 3GPP-style path loss model and complex channel gain generation
for a multi-user OFDMA downlink system. All formulas follow the paper's equations (5)–(8).
作者/Author: Sisyphus-Junior
日期/Date: 2026-02-28
论文引用/Paper Reference: Co-MADDPG based Resource Allocation for Semantic Communication
依赖/Dependencies: numpy
"""
import numpy as np
class ChannelModel:
"""
多用户 OFDMA 系统的频率选择性信道模型。
Frequency-selective channel model for multi-user OFDMA systems.
生成包含距离相关路径损耗和瑞利衰落的每子载波复信道增益，并计算每子载波的信噪比 (SNR)。
Generates per-subcarrier complex channel gains incorporating distance-dependent
path loss with Rayleigh fading, and computes per-subcarrier SNR values.
Parameters
----------
config : dict
完整的配置字典（必须包含 "env" 部分，且具有 carrier_freq, noise_psd 和 subcarrier_spacing 键）。
Full configuration dictionary (must contain an "env" section with keys
"carrier_freq", "noise_psd", and "subcarrier_spacing").
"""
def __init__(self, config: dict) -> None:
# 初始化环境配置 / Initialize environment configurations
self.config = config
env = config["env"]
# 载波频率 (GHz) / Carrier frequency in GHz
self._carrier_freq_ghz: float = env["carrier_freq"]
# 噪声功率谱密度 (dBm/Hz) / Noise power spectral density in dBm/Hz
self._noise_psd_dbm: float = env["noise_psd"]
# 子载波间隔 (Hz) / Subcarrier spacing in Hz
self._subcarrier_spacing: float = env["subcarrier_spacing"]
# ------------------------------------------------------------------
# 路径损耗 / Path loss
# ------------------------------------------------------------------
def path_loss(self, distance: float) -> float:
"""
计算与距离相关的路径损耗 (dB)。
Compute distance-dependent path loss in dB.
使用 3GPP Urban Micro (UMi) NLOS 模型 (公式 5):
Uses the 3GPP Urban Micro (UMi) NLOS model (Eq. 5):
PL(d) = 36.7 * log10(d) + 22.7 + 26 * log10(fc)
其中 d 的单位为米，fc 的单位为 GHz。
where *d* is in metres and *fc* is in GHz.
Parameters
----------
distance : float or np.ndarray
收发机之间的距离，单位为米。
Transmitter–receiver distance(s) in metres.
Returns
-------
float or np.ndarray
路径损耗值，单位为 dB。
Path loss value(s) in dB.
"""
fc = self._carrier_freq_ghz
# 应用 3GPP UMi NLOS 公式 / Apply 3GPP UMi NLOS formula - Eq.(5)
return 36.7 * np.log10(distance) + 22.7 + 26.0 * np.log10(fc)
# ------------------------------------------------------------------
# 信道生成 / Channel generation
# ------------------------------------------------------------------
def generate_channel(
self, distances: np.ndarray, num_subcarriers: int
) -> np.ndarray:
"""
生成所有用户和子载波的复信道增益。
Generate complex channel gains for all users and subcarriers.
每个元素 h_{k,n} 服从复高斯分布 CN(0, 10^{-PL/10}) (公式 6)。
即：独立循环对称复高斯分布，其方差等于线性尺度的逆路径损耗。
Each element h_{k,n} is drawn from CN(0, 10^{-PL/10}) (Eq. 6), i.e.
independent circularly-symmetric complex Gaussian with variance
equal to the linear-scale inverse path loss.
Parameters
----------
distances : array_like, shape (K,)
每个用户距离基站的距离（米）。
Distance of each user from the base station (metres).
num_subcarriers : int
OFDM 子载波数量 N。
Number of OFDM subcarriers *N*.
Returns
-------
np.ndarray, shape (K, N)
复信道增益矩阵。
Complex channel gain matrix.
"""
distances = np.asarray(distances, dtype=np.float64)
K = len(distances)
N = num_subcarriers
# 每用户路径损耗 -> 线性尺度信道方差 / Per-user path loss -> linear-scale channel variance
pl_db = self.path_loss(distances) # (K,)
# 方差 = 10^(-PL/10) / Variance = 10^(-PL/10) - Eq.(6)
variance = 10.0 ** (-pl_db / 10.0) # (K,)
variance = variance.reshape(K, 1) # (K, 1) 用于广播 / for broadcasting
# 复高斯：每个分量服从 N(0, var/2) / Complex Gaussian: each component ~ N(0, var/2)
std = np.sqrt(variance / 2.0)
# 生成实部和虚部 / Generate real and imaginary parts
real_part = np.random.randn(K, N) * std
imag_part = np.random.randn(K, N) * std
# 返回复增益 / Return complex gains
return real_part + 1j * imag_part
# ------------------------------------------------------------------
# SNR 计算 / SNR computation
# ------------------------------------------------------------------
def compute_snr(
self,
channel_gains: np.ndarray,
power_alloc: np.ndarray,
noise_power: float,
) -> np.ndarray:
"""
计算每个用户的每子载波信噪比 (SNR)。
Compute per-subcarrier SNR for every user.
γ_{k,n} = p_{k,n} * |h_{k,n}|² / σ² (公式 8)
γ_{k,n} = p_{k,n} · |h_{k,n}|² / σ² (Eq. 8)
Parameters
----------
channel_gains : np.ndarray, shape (K, N)
复信道增益矩阵。
Complex channel gain matrix.
power_alloc : np.ndarray, shape (K, N)
每个用户在每个子载波上分配的功率（瓦特）。
Power allocated by each user on each subcarrier (Watts).
noise_power : float
每子载波的噪声功率 σ²（瓦特）。
Noise power σ² per subcarrier (Watts).
Returns
-------
np.ndarray, shape (K, N)
SNR 值（线性尺度）。
SNR values (linear scale).
"""
# 计算 SNR = 功率 * 增益平方 / 噪声 / Compute SNR = Power * Gain^2 / Noise - Eq.(8)
return power_alloc * (np.abs(channel_gains) ** 2) / noise_power
# ------------------------------------------------------------------
# 噪声功率属性 / Noise power property
# ------------------------------------------------------------------
@property
def noise_power(self) -> float:
"""
每子载波的热噪声功率 (瓦特)。
Thermal noise power per subcarrier (Watts).
σ² = N₀ * Δf (公式 7)
σ² = N₀ · Δf (Eq. 7)
其中 N₀ 是从 dBm/Hz 转换为线性 (W/Hz) 的噪声功率谱密度：
where N₀ is the noise PSD converted from dBm/Hz to linear (W/Hz):
N₀_linear = 10^((N₀_dBm - 30) / 10)
Returns
-------
float
噪声功率（瓦特）。
Noise power in Watts.
"""
n0_dbm = self._noise_psd_dbm
delta_f = self._subcarrier_spacing
# 转换为线性功率谱密度 / Convert to linear PSD - Eq.(7)
n0_linear = 10.0 ** ((n0_dbm - 30.0) / 10.0)
# 计算总噪声功率 / Compute total noise power
return n0_linear * delta_f
code/envs/semantic_module.py
+156
View File
@@ -0,0 +1,156 @@
"""
语义通信模块 / Semantic communication module for Co-MADDPG.
实现基于 DeepSC 经验曲线的语义相似度 (SSim) 计算，以及语义通信用户的 QoE 计算。
Implements semantic similarity (SSim) computation based on empirical DeepSC curves,
and QoE calculation for semantic communication users.
作者/Author: Sisyphus-Junior
日期/Date: 2026-02-28
论文引用/Paper Reference: Co-MADDPG based Resource Allocation for Semantic Communication
依赖/Dependencies: numpy
"""
import numpy as np
class SemanticModule:
"""
语义通信质量模块。
Semantic communication quality module.
根据平均 SNR 和压缩率，使用受 DeepSC 文献启发的经验拟合曲线计算语义相似度指数 (SSim)。
Computes semantic similarity index (SSim) from average SNR and compression
ratio, using empirical fitting curves inspired by DeepSC literature.
Parameters
----------
config : dict
完整的配置字典（必须包含 "env" 部分，且具有 rho_max, rho_min, w1, w2 键）。
Full configuration dictionary (must contain an "env" section
with keys "rho_max", "rho_min", "w1", "w2").
"""
def __init__(self, config: dict) -> None:
# 初始化语义参数 / Initialize semantic parameters
self.config = config
env = config["env"]
# 最大压缩率 / Maximum compression ratio ρ_max
self.rho_max = env.get("rho_max", 1.0)
# 最小压缩率 / Minimum compression ratio ρ_min
self.rho_min = env.get("rho_min", 0.05)
# QoE 权重 1 / QoE Weight w1 (SSim weight)
self.w1 = env.get("w1", 0.7)
# QoE 权重 2 / QoE Weight w2 (Resource efficiency weight)
self.w2 = env.get("w2", 0.3)
@staticmethod
def _a(rho: float) -> float:
"""经验曲线参数 a(ρ) = 0.8 / (ρ + 0.1)。 / Empirical curve parameter a(ρ) = 0.8 / (ρ + 0.1)."""
return 0.8 / (rho + 0.1)
@staticmethod
def _b(rho: float) -> float:
"""经验曲线参数 b(ρ) = 0.6 + 0.2 * ρ。 / Empirical curve parameter b(ρ) = 0.6 + 0.2 * ρ."""
return 0.6 + 0.2 * rho
def compute_ssim(self, avg_snr, rho: float):
"""
计算语义相似度指数 (SSim)。
Compute semantic similarity index (SSim).
φ(γ̄, ρ) = 1 - exp(-a(ρ) * γ̄^{b(ρ)}) (公式参考 SSim 章节)
φ(γ̄, ρ) = 1 - exp(-a(ρ) * γ̄^{b(ρ)}) (Refer to SSim section in paper)
Parameters
----------
avg_snr : float or np.ndarray
线性尺度的平均 SNR（非 dB）。
Average SNR in linear scale (not dB).
rho : float
压缩率 ρ ∈ [ρ_min, ρ_max]。
Compression ratio ρ ∈ [ρ_min, ρ_max].
Returns
-------
float or np.ndarray
[0, 1] 范围内的语义相似度。
Semantic similarity in [0, 1].
"""
# 防止 SNR 过小导致数值错误 / Avoid numerical errors with small SNR
avg_snr = np.maximum(avg_snr, 1e-10)
# 获取经验参数 a 和 b / Get empirical parameters a and b
a = self._a(rho)
b = self._b(rho)
# 计算 SSim 公式 / Compute SSim formula
return 1.0 - np.exp(-a * np.power(avg_snr, b))
def compute_avg_snr(self, snr_per_subcarrier: np.ndarray,
allocation_mask: np.ndarray) -> float:
"""
计算已分配子载波上的平均 SNR。
Compute average SNR over allocated subcarriers.
Parameters
----------
snr_per_subcarrier : np.ndarray
所有子载波的 SNR 值（线性尺度）。
SNR values for all subcarriers (linear scale).
allocation_mask : np.ndarray
指示已分配子载波的二进制掩码。
Binary mask indicating allocated subcarriers.
Returns
-------
float
已分配子载波的平均 SNR；若未分配则返回 0。
Mean SNR over allocated subcarriers; 0 if none allocated.
"""
# 提取已分配子载波的 SNR / Extract SNR for allocated subcarriers
allocated = snr_per_subcarrier[allocation_mask > 0]
# 如果没有子载波被分配 / If no subcarriers are allocated
if len(allocated) == 0:
return 0.0
# 返回平均值 / Return mean value
return float(np.mean(allocated))
def compute_semantic_qoe(self, ssim: float, rho: float,
w1: float = None, w2: float = None,
rho_max: float = None) -> float:
"""
计算语义通信用户的 QoE。
Compute QoE for a semantic communication user.
QoE_s = w1 * SSim + w2 * (1 - ρ / ρ_max) (公式参考 QoE_s)
QoE_s = w1 * SSim + w2 * (1 - ρ / ρ_max) (Refer to QoE_s formula)
Parameters
----------
ssim : float
[0, 1] 范围内的语义相似度指数。
Semantic similarity index in [0, 1].
rho : float
使用的压缩率。
Compression ratio used.
w1, w2 : float, optional
权重（默认为配置中的实例值）。
Weights (defaults to instance values from config).
rho_max : float, optional
最大压缩率（默认为配置中的值）。
Maximum compression ratio (default from config).
Returns
-------
float
[0, 1] 范围内的 QoE 值。
QoE value in [0, 1].
"""
# 使用默认值或输入值 / Use default or input values
if w1 is None:
w1 = self.w1
if w2 is None:
w2 = self.w2
if rho_max is None:
rho_max = self.rho_max
# 计算语义 QoE / Calculate semantic QoE
return float(w1 * ssim + w2 * (1.0 - rho / rho_max))
code/envs/wireless_env.py
+336
View File
@@ -0,0 +1,336 @@
"""
无线资源分配环境 / Main Gym-like environment for wireless resource allocation.
该模块实现了一个用于语义和传统用户共存系统的无线资源分配环境。
它通过 Gym 风格的 reset/step 接口，处理子载波分配、功率控制和压缩率优化。
This module implements a wireless resource allocation environment for systems
with coexisting semantic and traditional users. It handles subcarrier allocation,
power control, and compression ratio optimization via a Gym-like reset/step interface.
作者/Author: Sisyphus-Junior
日期/Date: 2026-02-28
论文引用/Paper Reference: Co-MADDPG based Resource Allocation for Semantic Communication
依赖/Dependencies: numpy, envs.channel_model, envs.semantic_module
"""
import numpy as np
from envs.channel_model import ChannelModel
from envs.semantic_module import SemanticModule
class WirelessEnv:
"""
语义与传统通信共存环境。
Wireless environment with semantic and traditional communication.
管理信道状态、执行动作并计算系统范围内的 QoE。
Manages channel states, executes actions, and computes system-wide QoE.
Parameters
----------
config : dict
包含 'env' 和 'training' 部分的配置字典。
Configuration dictionary containing 'env' and 'training' sections.
"""
def __init__(self, config):
# 提取环境和训练配置 / Extract environment and training configs
env_config = config['env']
train_config = config['training']
# 核心系统参数 / Core system parameters
self.N = env_config['num_subcarriers'] # 子载波数量 N / Number of subcarriers
self.K_s = env_config['num_semantic_users'] # 语义用户数 / Number of semantic users
self.K_b = env_config['num_traditional_users'] # 传统用户数 / Number of traditional users
self.K = self.K_s + self.K_b # 总用户数 / Total number of users
# 物理层参数 / Physical layer parameters
self.P_max = env_config['max_power'] # 最大总发射功率 / Maximum total transmit power
self.R_req = env_config['min_rate_req'] # 传统用户最小速率需求 / Min rate requirement for traditional users
self.delta_f = env_config['subcarrier_spacing'] # 子载波间隔 / Subcarrier spacing
self.rho_min = env_config['rho_min'] # 最小压缩率 / Minimum compression ratio
self.rho_max = env_config['rho_max'] # 最大压缩率 / Maximum compression ratio
self.w1 = env_config['w1'] # 语义 QoE 权重 1 / Semantic QoE weight 1
self.w2 = env_config['w2'] # 语义 QoE 权重 2 / Semantic QoE weight 2
# 距离限制 / Distance limits
self.min_d = env_config.get('min_distance', 50.0)
self.max_d = env_config.get('max_distance', 500.0)
# 训练步数控制 / Training step control
self.max_steps = train_config['max_steps']
self.step_count = 0
# 初始化模型 / Initialize models
self.channel_model = ChannelModel(config)
self.semantic_module = SemanticModule(config)
# 初始状态变量 / Initial state variables
self.distances = np.zeros(self.K) # 用户距离 / User distances
self.channel_gains = np.zeros((self.K, self.N), dtype=complex) # 复信道增益 / Complex channel gains
self.content_sensitivity = 0.5 # 内容敏感度 / Content sensitivity
self.business_priority = 0.5 # 业务优先级 / Business priority
self.load_s = 0.5 # 语义流量负载 / Semantic traffic load
self.load_b = 0.5 # 传统流量负载 / Traditional traffic load
self.alloc_s = 0.0 # 语义子载波分配比例 / Semantic subcarrier allocation fraction
self.alloc_b = 0.0 # 传统子载波分配比例 / Traditional subcarrier allocation fraction
self.qoe_avg_s = 0.0 # 语义平均 QoE / Rolling average semantic QoE
self.qoe_avg_b = 0.0 # 传统平均 QoE / Rolling average traditional QoE
@property
def obs_dim(self):
"""观察维度: 子载波 (N) + 4 个额外特征。 / Observation dimension: Subcarriers (N) + 4 extra features."""
return self.N + 4
@property
def act_dim(self):
"""动作维度: 子载波比例, 功率比例, [语义: 压缩率]。 / Action dimension: Subcarrier fraction, Power fraction, [Semantic: Compression ratio]."""
return 3
def reset(self):
"""
重置环境状态。
Reset environment state.
Returns
-------
tuple
(语义智能体观察, 传统智能体观察)。
(semantic_observation, traditional_observation).
"""
# 在 [min_distance, max_distance] 内随机分配用户距离 / Random user distances in [min_distance, max_distance]
self.distances = np.random.uniform(self.min_d, self.max_d, size=self.K)
# 生成信道 (形状: K x N 复数) / Generate channel (shape: K x N complex) - Eq.(6)
self.channel_gains = self.channel_model.generate_channel(self.distances, self.N)
self.step_count = 0
# 随机设置观察参数 / Random params for observation
self.content_sensitivity = np.random.uniform(0.3, 0.8)
self.business_priority = np.random.uniform(0.3, 0.8)
self.load_s = np.random.uniform(0.2, 0.8)
self.load_b = np.random.uniform(0.2, 0.8)
# 重置分配比例和移动平均值 / Reset allocations and moving averages
self.alloc_s = 0.0
self.alloc_b = 0.0
self.qoe_avg_s = 0.0
self.qoe_avg_b = 0.0
# 获取初始观察 / Get initial observations
obs_s = self._get_observation('semantic')
obs_b = self._get_observation('traditional')
return obs_s, obs_b
def _get_observation(self, agent_type):
"""
构造智能体的观察向量。
Construct observation vector for agents.
Parameters
----------
agent_type : str
'semantic' 或 'traditional'。
'semantic' or 'traditional'.
Returns
-------
np.ndarray
归一化后的观察向量。
Normalized observation vector.
"""
if agent_type == 'semantic':
# 语义用户索引范围 / Semantic user indices range
user_indices = range(self.K_b, self.K)
if len(user_indices) > 0:
# 计算平均信道增益平方 (功率) / Mean channel power
channel_power = np.mean(np.abs(self.channel_gains[user_indices])**2, axis=0)
else:
channel_power = np.zeros(self.N)
# 归一化信道功率 / Normalize channel power
channel_norm = channel_power / (np.max(channel_power) + 1e-10)
# 拼接额外特征 / Concatenate extra features
obs = np.concatenate([channel_norm,
[self.qoe_avg_s, self.content_sensitivity, self.alloc_s, self.load_s]])
else: # 传统 / traditional
# 传统用户索引范围 / Traditional user indices range
user_indices = range(0, self.K_b)
if len(user_indices) > 0:
# 计算平均信道功率 / Mean channel power
channel_power = np.mean(np.abs(self.channel_gains[user_indices])**2, axis=0)
else:
channel_power = np.zeros(self.N)
# 归一化信道功率 / Normalize channel power
channel_norm = channel_power / (np.max(channel_power) + 1e-10)
# 拼接额外特征 / Concatenate extra features
obs = np.concatenate([channel_norm,
[self.qoe_avg_b, self.business_priority, self.alloc_b, self.load_b]])
# 返回 32位浮点型观察 / Return float32 observation
return obs.astype(np.float32)
def step(self, action_s, action_b):
"""
执行一个时间步。
Execute a single environment step.
Parameters
----------
action_s : np.ndarray
语义智能体动作 [子载波比例, 功率比例, 压缩率]。
Semantic agent action [sub_fraction, power_fraction, compression_ratio].
action_b : np.ndarray
传统智能体动作 [子载波比例, 功率比例, 冗余参数]。
Traditional agent action [sub_fraction, power_fraction, redundant_param].
Returns
-------
tuple
(obs_s, obs_b, reward_s, reward_b, done, info).
"""
self.step_count += 1
# 1. 解码动作 / Decode actions
# 计算子载波分配数量 / Compute number of subcarriers
n_sub_s = max(1, int(round(action_s[0] * self.N)))
n_sub_b = max(1, int(round(action_b[0] * self.N)))
# 限制总子载波数量 / Clip total subcarriers
if n_sub_s + n_sub_b > self.N:
total = n_sub_s + n_sub_b
n_sub_s = int(round(n_sub_s * self.N / total))
n_sub_b = self.N - n_sub_s
# 计算功率分配 / Compute power allocation
p_s = action_s[1] * self.P_max
p_b = action_b[1] * self.P_max
# 限制总功率 / Limit total power
if p_s + p_b > self.P_max:
total_p = p_s + p_b
p_s = p_s * self.P_max / total_p
p_b = p_b * self.P_max / total_p
# 解码语义压缩率 / Decode semantic compression ratio
rho = action_s[2] * (self.rho_max - self.rho_min) + self.rho_min
# 2. 分配子载波 (基于信道质量的贪婪算法) / Allocate subcarriers (greedy by channel quality)
# 计算两组用户的平均信道质量 / Mean channel quality for both groups
sem_channel = np.mean(np.abs(self.channel_gains[self.K_b:])**2, axis=0) if self.K_s > 0 else np.zeros(self.N)
trad_channel = np.mean(np.abs(self.channel_gains[:self.K_b])**2, axis=0) if self.K_b > 0 else np.zeros(self.N)
# 语义用户优先挑选最好的子载波 / Semantic users pick best subcarriers first
all_subs = np.arange(self.N)
sem_sorted = np.argsort(-sem_channel)
sem_subs = sem_sorted[:n_sub_s]
# 剩余子载波给传统用户 / Remaining subcarriers for traditional users
remaining = np.setdiff1d(all_subs, sem_subs)
if len(remaining) >= n_sub_b:
trad_quality = trad_channel[remaining]
best_idx = np.argsort(-trad_quality)[:n_sub_b]
trad_subs = remaining[best_idx]
else:
trad_subs = remaining
n_sub_b = len(trad_subs)
# 3. 功率分配 (组内均分) / Power allocation (equal within group)
noise_power = self.channel_model.noise_power
# 分配矩阵和功率矩阵 / Allocation and power matrices
alloc_matrix = np.zeros((self.K, self.N))
power_matrix = np.zeros((self.K, self.N))
# 在 K_s 个用户中循环分配语义子载波 / Distribute semantic subcarriers among K_s users round-robin
for i, k in enumerate(range(self.K_b, self.K)):
user_subs = sem_subs[i::max(1, self.K_s)]
if len(user_subs) > 0:
alloc_matrix[k, user_subs] = 1
power_matrix[k, user_subs] = p_s / max(n_sub_s, 1)
# 在 K_b 个用户中循环分配传统子载波 / Distribute traditional subcarriers among K_b users round-robin
for i, k in enumerate(range(0, self.K_b)):
user_subs = trad_subs[i::max(1, self.K_b)]
if len(user_subs) > 0:
alloc_matrix[k, user_subs] = 1
power_matrix[k, user_subs] = p_b / max(n_sub_b, 1)
# 4. 计算 SNR / Compute SNR - Eq.(8)
snr_matrix = self.channel_model.compute_snr(self.channel_gains, power_matrix, noise_power)
# 5. 计算每个用户的 QoE / Compute QoE for each user
qoe_list = []
rates = []
ssim_values = []
# 传统用户 QoE 计算 / Traditional users QoE computation - Eq.(QoE_b)
for k in range(self.K_b):
user_subs = np.where(alloc_matrix[k] > 0)[0]
if len(user_subs) == 0:
rate_k = 0.0
else:
# R_k = Σ α * Δf * log2(1 + γ) / R_k = Σ α * Δf * log2(1 + γ)
rate_k = np.sum(self.delta_f * np.log2(1 + snr_matrix[k, user_subs]))
rates.append(rate_k)
# QoE_b = min(R_k / R_req, 1) / QoE_b = min(R_k / R_req, 1)
qoe_k = min(rate_k / self.R_req, 1.0)
qoe_list.append(qoe_k)
# 语义用户 QoE 计算 / Semantic users QoE computation - Eq.(QoE_s)
for k in range(self.K_b, self.K):
user_subs = np.where(alloc_matrix[k] > 0)[0]
if len(user_subs) == 0:
ssim_k = 0.0
else:
avg_snr = np.mean(snr_matrix[k, user_subs])
# 计算语义相似度 / Compute SSim - Eq. (SSim)
ssim_k = self.semantic_module.compute_ssim(avg_snr, rho)
ssim_values.append(float(ssim_k))
# 计算语义 QoE / Compute semantic QoE
qoe_k = self.semantic_module.compute_semantic_qoe(ssim_k, rho, self.w1, self.w2, self.rho_max)
qoe_list.append(qoe_k)
# 6. 系统平均 QoE / System QoE
qoe_sys = np.mean(qoe_list) if len(qoe_list) > 0 else 0.0
qoe_s = np.mean(qoe_list[self.K_b:]) if self.K_s > 0 else 0.0
qoe_b = np.mean(qoe_list[:self.K_b]) if self.K_b > 0 else 0.0
# 更新滚动平均值 / Update rolling averages
alpha_smooth = 0.1
self.qoe_avg_s = alpha_smooth * qoe_s + (1 - alpha_smooth) * self.qoe_avg_s
self.qoe_avg_b = alpha_smooth * qoe_b + (1 - alpha_smooth) * self.qoe_avg_b
# 记录当前分配比例 / Record current allocation ratios
self.alloc_s = n_sub_s / self.N
self.alloc_b = n_sub_b / self.N
# 7. 为下一步生成新信道 (块衰落) / Regenerate channel for next step (block fading)
self.channel_gains = self.channel_model.generate_channel(self.distances, self.N)
# 8. 构造输出数据 / Build output
obs_s = self._get_observation('semantic')
obs_b = self._get_observation('traditional')
done = (self.step_count >= self.max_steps)
# 计算速率满足度 / Compute rate satisfaction for traditional users
if len(rates) > 0:
rate_satisfaction = float(np.mean([1.0 if r >= self.R_req else 0.0 for r in rates]))
else:
rate_satisfaction = 1.0
# 构造信息字典 / Construct info dictionary
info = {
'qoe_semantic': qoe_s,
'qoe_traditional': qoe_b,
'qoe_sys': qoe_sys,
'qoe_list': qoe_list,
'rates': rates,
'ssim_values': ssim_values,
'rate_satisfaction': rate_satisfaction,
'rho': rho,
'n_sub_s': n_sub_s,
'n_sub_b': n_sub_b,
}
# 返回结果 (奖励值设为各自的平均 QoE) / Return results (rewards set to respective mean QoEs)
return obs_s, obs_b, qoe_s, qoe_b, done, info