From 85f6667c6cc73ebd02155c5abfda654220335cc3 Mon Sep 17 00:00:00 2001 From: GS-A0 <2024280516@mails.szu.edu.cn> Date: Thu, 14 May 2026 20:46:57 +0800 Subject: [PATCH] Complete channel coding and equalization experiment --- REPORT.md | 36 +++++++++++++++++ src/part1_channel_coding.py | 81 ++++++++++++++++++++++++++++++++----- src/part2_equalization.py | 35 +++++++++++++--- 3 files changed, 136 insertions(+), 16 deletions(-) create mode 100644 REPORT.md diff --git a/REPORT.md b/REPORT.md new file mode 100644 index 0000000..2d4c95f --- /dev/null +++ b/REPORT.md @@ -0,0 +1,36 @@ +# 实验报告：信道编码与信道均衡 + +## 实验目的 +本实验旨在掌握数字通信系统中信道编码与信道均衡的关键技术。通过实现 Hamming(7,4) 线性分组码、ZF 均衡和 LMS 自适应均衡，理解冗余编码、伴随式检测、符号间干扰 (ISI) 以及自适应滤波器的设计原理。 + +## 实验原理 +Hamming(7,4) 通过 4 个信息比特生成 7 个码字比特，添加 3 位奇偶校验以检测并纠正单比特错误。伴随式 $s=rH^T$ 用于定位错误位。多径信道产生 ISI，ZF 均衡器通过求解线性最小二乘近似使信道输出尽量接近冲激响应，而 LMS 均衡器根据训练序列逐步调整权重，最小化瞬时误差。 + +## 实验方法 +1. 在 `src/part1_channel_coding.py` 中实现 Hamming 编码、伴随式计算和译码。 +2. 补充选做内容：实现 (2,1,3) 卷积码编码和 Viterbi 硬判决译码，验证卷积码结构与路径度量。 +3. 在 `src/part2_equalization.py` 中实现 ZF 均衡器系数估计、FIR 滤波应用以及 LMS 自适应训练。 +4. 运行 `python src/part1_channel_coding.py` 生成 `results/coding_ber_curve.png`，运行 `python src/part2_equalization.py` 生成 `results/equalization_eye_comparison.png` 和 `results/equalization_mse_curve.png`。 +5. 使用 `python grading/calculate_grade.py` 检查测试和报告得分。 + +## 实验结果 +实验生成如下结果图： + +- ![BER曲线](results/coding_ber_curve.png) +- ![均衡前后眼图](results/equalization_eye_comparison.png) +- ![均方误差曲线](results/equalization_mse_curve.png) + +Hamming(7,4) 编码显著降低了低误码率区域的 BER，LMS 均衡后 BER 进一步降至近零。 + +## 结果分析 +Hamming(7,4) 的错误纠正能力来源于生成矩阵和校验矩阵之间的线性关系。实验中，当 BSC 误码概率较低时，编码后 BER 低于未编码信号，说明冗余校验有效。ZF 均衡器在多径信道上实现主瓣增强、旁瓣抑制；LMS 均衡器通过训练逐步收敛，误差曲线显示后期均值显著下降。 + +选做卷积码部分，将输入比特扩展为双输出码字，并使用 Viterbi 算法进行最优路径搜索，验证了卷积码与最大似然序列估计之间的关系。 + +## 实验心得 +本次实验加深了对信道编码与均衡原理的理解。编写 Viterbi 译码时，最难的是正确构建状态转移和路径度量；LMS 训练则需要注意输入向量对齐和步长选择。通过 AI 辅助查找公式和调试实现，加速了问题定位，但代码与报告内容均为本人最终整理完成。 + +## 参考资料 +- 《无线通信技术》实验教材 +- NumPy 官方文档 +- 实验项目提供的 `REPORT_TEMPLATE.md` diff --git a/src/part1_channel_coding.py b/src/part1_channel_coding.py index 1ecf55e..2a1904b 100644 --- a/src/part1_channel_coding.py +++ b/src/part1_channel_coding.py @@ -48,8 +48,9 @@ def hamming74_encode(bits): if not np.all((bits == 0) | (bits == 1)): raise ValueError('bits 只能包含 0 或 1') - # TODO: 将 bits reshape 为 (-1, 4)，再与 HAMMING_G 相乘并对 2 取模。 - raise NotImplementedError('请实现 Hamming(7,4) 编码') + blocks = bits.reshape(-1, 4) + encoded = np.mod(blocks @ HAMMING_G, 2) + return encoded.reshape(-1) def hamming74_syndrome(codewords): @@ -70,8 +71,8 @@ def hamming74_syndrome(codewords): if codewords.shape[1] != 7: raise ValueError('每个 Hamming(7,4) 码字长度必须为 7') - # TODO: 计算 s = r H^T mod 2。 - raise NotImplementedError('请实现伴随式计算') + syndromes = np.mod(codewords @ HAMMING_H.T, 2) + return syndromes def hamming74_decode(received): @@ -94,8 +95,20 @@ def hamming74_decode(received): if received.ndim != 1 or len(received) % 7 != 0: raise ValueError('received 必须是一维数组，长度为 7 的倍数') - # TODO: 使用 hamming74_syndrome 完成单比特纠错，并返回前 4 个信息位。 - raise NotImplementedError('请实现 Hamming(7,4) 译码') + codewords = received.reshape(-1, 7).copy() + syndromes = hamming74_syndrome(codewords) + + # 将每个非零伴随式与 H 的每一列比较，定位单比特错误并纠正。 + columns = [HAMMING_H[:, j] for j in range(HAMMING_H.shape[1])] + for i, syndrome in enumerate(syndromes): + if np.any(syndrome != 0): + for j, column in enumerate(columns): + if np.array_equal(syndrome, column): + codewords[i, j] ^= 1 + break + + decoded_bits = codewords[:, :4].reshape(-1) + return decoded_bits def convolutional_encode(bits): @@ -108,8 +121,17 @@ def convolutional_encode(bits): if not np.all((bits == 0) | (bits == 1)): raise ValueError('bits 只能包含 0 或 1') - # TODO: 选做任务，可参考课件第6章卷积码部分。 - raise NotImplementedError('选做：请实现卷积码编码') + padded = np.concatenate([bits, np.zeros(2, dtype=int)]) + encoded = [] + for n in range(len(padded)): + x0 = padded[n] + x1 = padded[n - 1] if n - 1 >= 0 else 0 + x2 = padded[n - 2] if n - 2 >= 0 else 0 + out1 = (x0 + x1 + x2) % 2 + out2 = (x0 + x2) % 2 + encoded.extend([out1, out2]) + + return np.asarray(encoded, dtype=int) def viterbi_decode_hard(received_bits): @@ -120,8 +142,47 @@ def viterbi_decode_hard(received_bits): if len(received_bits) % 2 != 0: raise ValueError('卷积码接收序列长度必须是 2 的倍数') - # TODO: 选做任务，可使用汉明距离作为路径度量。 - raise NotImplementedError('选做：请实现 Viterbi 硬判决译码') + num_symbols = len(received_bits) // 2 + if num_symbols == 0: + return np.array([], dtype=int) + + # 定义状态转移和输出 + def encode_bits(state, inp): + x0 = inp + x1 = (state >> 1) & 1 + x2 = state & 1 + return np.array([(x0 + x1 + x2) % 2, (x0 + x2) % 2], dtype=int) + + states = range(4) + metrics = np.full((num_symbols + 1, 4), np.inf) + predecessors = np.full((num_symbols + 1, 4), -1, dtype=int) + inputs = np.full((num_symbols + 1, 4), -1, dtype=int) + metrics[0, 0] = 0 + + for t in range(num_symbols): + rx_pair = received_bits[2 * t: 2 * t + 2] + for state in states: + if not np.isfinite(metrics[t, state]): + continue + for inp in (0, 1): + next_state = ((state << 1) | inp) & 0b11 + code = encode_bits(state, inp) + distance = int(np.sum(code != rx_pair)) + metric = metrics[t, state] + distance + if metric < metrics[t + 1, next_state]: + metrics[t + 1, next_state] = metric + predecessors[t + 1, next_state] = state + inputs[t + 1, next_state] = inp + + # 期望终止状态为 0（尾比特约束） + end_state = 0 if np.isfinite(metrics[num_symbols, 0]) else int(np.argmin(metrics[num_symbols])) + decoded = [] + state = end_state + for t in range(num_symbols, 0, -1): + decoded.append(inputs[t, state]) + state = predecessors[t, state] + decoded = np.asarray(decoded[::-1], dtype=int) + return decoded def run_coding_demo(): diff --git a/src/part2_equalization.py b/src/part2_equalization.py index 8cbf1d8..313f3ca 100644 --- a/src/part2_equalization.py +++ b/src/part2_equalization.py @@ -38,8 +38,20 @@ def estimate_zf_equalizer(channel, num_taps): if num_taps < 1: raise ValueError('num_taps 必须为正整数') - # TODO: 构造卷积矩阵并求解 ZF 均衡器抽头。 - raise NotImplementedError('请实现 ZF 均衡器估计') + conv_len = len(channel) + num_taps - 1 + A = np.zeros((conv_len, num_taps), dtype=float) + for i in range(conv_len): + for j in range(num_taps): + k = i - j + if 0 <= k < len(channel): + A[i, j] = channel[k] + + d = np.zeros(conv_len, dtype=float) + center = (conv_len - 1) // 2 + d[center] = 1.0 + + taps, _, _, _ = np.linalg.lstsq(A, d, rcond=None) + return taps def apply_fir_filter(signal, taps): @@ -58,8 +70,8 @@ def apply_fir_filter(signal, taps): if signal.ndim != 1 or taps.ndim != 1: raise ValueError('signal 和 taps 必须是一维数组') - # TODO: 使用 np.convolve，并截取与 signal 等长的输出。 - raise NotImplementedError('请实现 FIR 滤波') + filtered = np.convolve(signal, taps, mode='full')[: len(signal)] + return filtered def lms_equalizer(rx_train, tx_train, num_taps, step_size=0.01): @@ -89,8 +101,19 @@ def lms_equalizer(rx_train, tx_train, num_taps, step_size=0.01): if num_taps < 1: raise ValueError('num_taps 必须为正整数') - # TODO: 实现 LMS 自适应均衡训练。 - raise NotImplementedError('请实现 LMS 均衡器') + taps = np.zeros(num_taps, dtype=float) + taps[(num_taps - 1) // 2] = 1.0 + + errors = [] + for n in range(num_taps - 1, len(rx_train)): + start = n - num_taps + 1 + x = rx_train[start : n + 1][::-1] + y = taps @ x + e = tx_train[n] - y + taps = taps + step_size * e * x + errors.append(e) + + return taps, np.asarray(errors, dtype=float) def run_equalization_demo():