import math import time import numpy as np import tkinter as tk from tkinter import ttk, messagebox, filedialog import torch import matplotlib as mpl from matplotlib.figure import Figure from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg, NavigationToolbar2Tk from matplotlib import font_manager # ========================================================= # 0. 全局设置 # ========================================================= def configure_matplotlib_fonts(): """自动选择可用中文字体，修复中文显示问题。""" candidate_fonts = [ "Microsoft YaHei", "SimHei", "Noto Sans CJK SC", "Noto Serif CJK SC", "WenQuanYi Zen Hei", "AR PL UMing CN", "PingFang SC", "Heiti SC", "STHeiti", "Songti SC", ] installed = {f.name for f in font_manager.fontManager.ttflist} selected = None for name in candidate_fonts: if name in installed: selected = name break mpl.rcParams["axes.unicode_minus"] = False mpl.rcParams["mathtext.fontset"] = "stix" if selected: mpl.rcParams["font.sans-serif"] = [selected] else: mpl.rcParams["font.sans-serif"] = ["DejaVu Sans"] configure_matplotlib_fonts() device = torch.device("cuda" if torch.cuda.is_available() else "cpu") torch.manual_seed(2024) np.random.seed(2024) L_SNAPSHOTS_DEFAULT = 100 MC_DEFAULT = 100 # 颜色表：修复 METHOD_COLORS 未定义 METHOD_COLORS = { "Analog": "#77AC30", "Digital": "#000000", "DFT": "#4DBEEE", "OMP": "#D95319", "TSC-OMP": "#0072BD", } # ========================================================= # 1. 算法引擎 # ========================================================= class BeamEngine: def __init__(self): self.device = device self.N_ant = 24 self.N_RF = 4 self.noise_power = 1.0 self.INR_linear = 10 ** (30 / 10) # ---------- 阵列模型 ---------- def get_positions_lambda(self, array_type: str, n_ant: int): array_type = array_type.upper() if array_type == "ULA": pos = 0.5 * np.arange(-(n_ant - 1) / 2, (n_ant - 1) / 2 + 1) return pos.astype(np.float32) # 24 阵元 NULA：沿用现有主线代码 if array_type == "NULA" and n_ant == 24: fc = 4.9e9 lam = 3e8 / fc a = np.arange(0, 11 * 43 + 1, 43) b = np.array([11 * 43 + 63]) c = 11 * 43 + 63 + np.arange(43, 11 * 43 + 1, 43) delta_d = 1e-3 * np.concatenate((a, b, c)) pos = delta_d / lam return pos.astype(np.float32) # 兜底广义 NULA base = 0.5 * np.arange(n_ant) jitter = 0.08 * np.sin(np.linspace(0, 3 * np.pi, n_ant)) pos = base + jitter pos = pos - np.mean(pos) return pos.astype(np.float32) def gen_a(self, theta_deg, array_type="ULA"): pos = self.get_positions_lambda(array_type, self.N_ant) pos_t = torch.tensor(pos, dtype=torch.float32, device=self.device).unsqueeze(1) theta_rad = torch.deg2rad(torch.tensor(theta_deg, dtype=torch.float32, device=self.device)) if array_type.upper() == "ULA": return torch.exp(1j * 2 * torch.pi * pos_t * torch.sin(theta_rad)) return torch.exp(-1j * 2 * torch.pi * pos_t * torch.sin(theta_rad)) def build_dictionary(self, angle_grid, array_type="ULA"): return torch.cat( [self.gen_a(float(th), array_type) / math.sqrt(self.N_ant) for th in angle_grid], dim=1 ) # ---------- 子程序 ---------- def extract_omp(self, w_opt, a_dic): res = w_opt.clone() f_rf = None for _ in range(self.N_RF): proj = a_dic.mH @ res idx = torch.argmax(torch.abs(proj)) atom = a_dic[:, idx:idx + 1] f_rf = atom if f_rf is None else torch.cat((f_rf, atom), dim=1) res = w_opt - f_rf @ (torch.linalg.pinv(f_rf) @ w_opt) return f_rf def make_prior_broad(self, theta_prior, window_half_width, array_type="ULA"): a_prior_broad = torch.zeros((self.N_ant, 1), dtype=torch.complex64, device=self.device) for tb in torch.arange(theta_prior - window_half_width, theta_prior + window_half_width + 0.1, 0.5, device=self.device): a_prior_broad += self.gen_a(float(tb), array_type) a_prior_broad /= torch.norm(a_prior_broad) return a_prior_broad def calc_hybrid_weight(self, f_rf, a_target, r_in): a_eff = f_rf.mH @ a_target r_eff = f_rf.mH @ r_in @ f_rf r_eff = r_eff + (0.05 * torch.trace(r_eff).real / self.N_RF) * torch.eye(self.N_RF, device=self.device) inv_r_eff = torch.linalg.inv(r_eff) f_bb = (inv_r_eff @ a_eff) / (a_eff.mH @ inv_r_eff @ a_eff) w = f_rf @ f_bb return w / torch.norm(w) def build_all_structures(self, target_angle, clutter_angle, array_type="ULA", window_half_width=5.0): theta_prior = round(target_angle) a_target = self.gen_a(target_angle, array_type) a_clutter = self.gen_a(clutter_angle, array_type) # 干扰+噪声先验协方差 r_interference_prior = ( self.noise_power * torch.eye(self.N_ant, device=self.device) + self.INR_linear * (a_clutter @ a_clutter.mH) ) a_prior_broad = self.make_prior_broad(theta_prior, window_half_width, array_type) w_digital_prior = torch.linalg.inv(r_interference_prior) @ a_prior_broad w_digital_prior /= torch.norm(w_digital_prior) # 传统 OMP / TSC-OMP 字典 dic_angles_full = np.arange(-90, 90.5, 0.5) dic_angles_tsc = np.arange( theta_prior - window_half_width, theta_prior + window_half_width + 0.1, 0.1 ) a_dic_full = self.build_dictionary(dic_angles_full, array_type) a_dic_tsc = self.build_dictionary(dic_angles_tsc, array_type) # DFT 射频矩阵 angles_dft = torch.linspace( theta_prior - window_half_width, theta_prior + window_half_width, self.N_RF, device=self.device ) f_rf_dft = torch.cat( [self.gen_a(float(th), array_type) / math.sqrt(self.N_ant) for th in angles_dft], dim=1 ) f_rf_omp = self.extract_omp(w_digital_prior, a_dic_full) f_rf_tsc = self.extract_omp(w_digital_prior, a_dic_tsc) # 静态理想协方差，用于方向图和纯数字权值 sig_power_static = 10 ** (0 / 10) r_in_static = ( self.noise_power * torch.eye(self.N_ant, device=self.device) + sig_power_static * (a_target @ a_target.mH) + self.INR_linear * (a_clutter @ a_clutter.mH) ) r_in_static_dl = r_in_static + ( 0.01 * torch.trace(r_in_static).real / self.N_ant ) * torch.eye(self.N_ant, device=self.device) w_analog = a_target / math.sqrt(self.N_ant) inv_r_stat = torch.linalg.inv(r_in_static_dl) w_digital = (inv_r_stat @ a_target) / (a_target.mH @ inv_r_stat @ a_target) w_digital /= torch.norm(w_digital) # DFT / OMP / TSC-OMP 混合 w_dft = self.calc_hybrid_weight(f_rf_dft, a_target, r_in_static) w_omp = self.calc_hybrid_weight(f_rf_omp, a_target, r_in_static) w_tsc = self.calc_hybrid_weight(f_rf_tsc, a_target, r_in_static) return { "theta_prior": theta_prior, "a_target": a_target, "a_clutter": a_clutter, "w_analog": w_analog, "w_digital": w_digital, "w_dft": w_dft, "w_omp": w_omp, "w_tsc": w_tsc, "f_rf_dft": f_rf_dft, "f_rf_omp": f_rf_omp, "f_rf_tsc": f_rf_tsc, } # ---------- 主功能1：波束方向图 ---------- def beam_patterns(self, target_angle, clutter_angle, array_type="ULA", methods=None, window_half_width=5.0): if methods is None: methods = ["Analog", "Digital", "DFT", "OMP", "TSC-OMP"] structures = self.build_all_structures(target_angle, clutter_angle, array_type, window_half_width) theta_scan = torch.arange(-90, 90.1, 0.1, device=self.device) a_scan = torch.cat([self.gen_a(float(th), array_type) for th in theta_scan], dim=1) weights = { "Analog": structures["w_analog"], "Digital": structures["w_digital"], "DFT": structures["w_dft"], "OMP": structures["w_omp"], "TSC-OMP": structures["w_tsc"], } result = {"theta_scan": theta_scan.cpu().numpy()} for method in methods: w = weights[method] gain = torch.abs(a_scan.mH @ w) ** 2 bp = 10 * torch.log10((gain / torch.max(gain)).squeeze()) result[method] = bp.cpu().numpy() return result # ---------- 主功能2：RMSE 曲线 ---------- def rmse_curves(self, target_angle, clutter_angle, array_type="ULA", methods=None, window_half_width=5.0, mc=100, snapshots=100, snr_values=None): """ RMSE 计算逻辑参考： - ULA: RMSE_Test_ula(2).py - NULA: RMSE_Test_nula(2).py 其中 DFT / OMP / TSC-OMP 均采用先验宽波束 + 低维 Capon 超分辨。 """ if methods is None: methods = ["Analog", "Digital", "DFT", "OMP", "TSC-OMP"] if snr_values is None: snr_values = np.arange(-10, 22, 2) structures = self.build_all_structures(target_angle, clutter_angle, array_type, window_half_width) theta_prior = structures["theta_prior"] a_target = structures["a_target"] a_clutter = structures["a_clutter"] scan_grid = torch.arange( theta_prior - window_half_width, theta_prior + window_half_width + 0.01, 0.01, device=self.device ) a_scan = torch.cat([self.gen_a(float(th), array_type) for th in scan_grid], dim=1) curves = {m: [] for m in methods} a_eff_dft = structures["f_rf_dft"].mH @ a_scan a_eff_omp = structures["f_rf_omp"].mH @ a_scan a_eff_tsc = structures["f_rf_tsc"].mH @ a_scan for snr in snr_values: sig_power = 10 ** (snr / 10) s_t = np.sqrt(sig_power / 2) * ( torch.randn(mc, 1, snapshots, device=self.device) + 1j * torch.randn(mc, 1, snapshots, device=self.device) ) s_c = np.sqrt(self.INR_linear / 2) * ( torch.randn(mc, 1, snapshots, device=self.device) + 1j * torch.randn(mc, 1, snapshots, device=self.device) ) n_t = np.sqrt(self.noise_power / 2) * ( torch.randn(mc, self.N_ant, snapshots, device=self.device) + 1j * torch.randn(mc, self.N_ant, snapshots, device=self.device) ) x = a_target.unsqueeze(0) * s_t + a_clutter.unsqueeze(0) * s_c + n_t r_in = torch.bmm(x, x.mH) / snapshots # 纯模拟：Bartlett if "Analog" in methods: p_ana = torch.einsum('si, bij, js -> bs', a_scan.conj().T, r_in, a_scan).abs() ang_ana = scan_grid[torch.argmax(p_ana, dim=1)] curves["Analog"].append(torch.sqrt(torch.mean((ang_ana - target_angle) ** 2)).item()) # 纯数字：全维 Capon / MVDR if "Digital" in methods: trace_r_in = torch.diagonal(r_in, dim1=-2, dim2=-1).sum(-1).real r_in_dl = r_in + ( 0.05 * trace_r_in / self.N_ant ).view(-1, 1, 1) * torch.eye(self.N_ant, device=self.device).unsqueeze(0) inv_r_in_dl = torch.linalg.inv(r_in_dl) p_dig = 1.0 / torch.abs(torch.einsum('si, bij, js -> bs', a_scan.conj().T, inv_r_in_dl, a_scan)) ang_dig = scan_grid[torch.argmax(p_dig, dim=1)] curves["Digital"].append(torch.sqrt(torch.mean((ang_dig - target_angle) ** 2)).item()) def process_hybrid(f_rf, a_eff): y = f_rf.mH.unsqueeze(0) @ x r = (y @ y.mH) / snapshots r = r + ( 0.05 * torch.diagonal(r, dim1=-2, dim2=-1).sum(-1).view(-1, 1, 1) / self.N_RF ) * torch.eye(self.N_RF, device=self.device).unsqueeze(0) inv_r = torch.linalg.inv(r) p = 1.0 / torch.abs(torch.einsum('si, bij, js -> bs', a_eff.conj().T, inv_r, a_eff)) ang = scan_grid[torch.argmax(p, dim=1)] return torch.sqrt(torch.mean((ang - target_angle) ** 2)).item() if "DFT" in methods: curves["DFT"].append(process_hybrid(structures["f_rf_dft"], a_eff_dft)) if "OMP" in methods: curves["OMP"].append(process_hybrid(structures["f_rf_omp"], a_eff_omp)) if "TSC-OMP" in methods: curves["TSC-OMP"].append(process_hybrid(structures["f_rf_tsc"], a_eff_tsc)) curves["snr"] = np.array(snr_values) return curves # ========================================================= # 2. 绘图面板 # ========================================================= class PlotPanel(ttk.Frame): def __init__(self, master): super().__init__(master) self.figure = Figure(figsize=(8.6, 5.8), dpi=100) self.canvas = FigureCanvasTkAgg(self.figure, master=self) self.canvas_widget = self.canvas.get_tk_widget() self.canvas_widget.pack(fill="both", expand=True) self.toolbar = NavigationToolbar2Tk(self.canvas, self, pack_toolbar=False) self.toolbar.update() self.toolbar.pack(fill="x") def redraw(self): self.canvas.draw_idle() def save_figure(self): path = filedialog.asksaveasfilename( defaultextension=".png", filetypes=[("PNG 图片", "*.png"), ("PDF 文件", "*.pdf"), ("SVG 文件", "*.svg")] ) if path: self.figure.savefig(path, dpi=300, bbox_inches="tight") return path return None # ========================================================= # 3. GUI 主体 # ========================================================= class HBFWorkbench(tk.Tk): def __init__(self): super().__init__() self.title("大规模阵列 MIMO 波束优化综合平台") self.geometry("1450x900") self.minsize(1200, 760) self.engine = BeamEngine() self._setup_style() self.status_var = tk.StringVar(value=f"就绪 | 设备: {device}") self._build_ui() def _setup_style(self): style = ttk.Style(self) try: style.theme_use("clam") except Exception: pass style.configure("TNotebook.Tab", padding=(14, 8), font=("Microsoft YaHei", 10)) style.configure("Header.TLabel", font=("Microsoft YaHei", 12, "bold")) style.configure("Accent.TButton", font=("Microsoft YaHei", 10, "bold")) def _build_ui(self): main = ttk.Frame(self) main.pack(fill="both", expand=True, padx=10, pady=10) left = ttk.Frame(main) left.pack(side="left", fill="y", padx=(0, 8)) right = ttk.Frame(main) right.pack(side="right", fill="both", expand=True) # ---------- 参数卡 ---------- card1 = ttk.LabelFrame(left, text="场景参数") card1.pack(fill="x", pady=(0, 8)) self.var_target = tk.DoubleVar(value=5.0) self.var_clutter = tk.DoubleVar(value=-15.0) self.var_window = tk.DoubleVar(value=5.0) self.var_mc = tk.IntVar(value=MC_DEFAULT) self.var_snap = tk.IntVar(value=L_SNAPSHOTS_DEFAULT) self.var_array = tk.StringVar(value="ULA") self.var_function = tk.StringVar(value="beam") self._add_labeled_entry(card1, "目标角度 (°)", self.var_target) self._add_labeled_entry(card1, "干扰角度 (°)", self.var_clutter) self._add_labeled_entry(card1, "波门半宽 (°)", self.var_window) self._add_labeled_entry(card1, "Monte Carlo", self.var_mc) self._add_labeled_entry(card1, "快拍数", self.var_snap) row = ttk.Frame(card1) row.pack(fill="x", pady=4) ttk.Label(row, text="阵列类型", width=12).pack(side="left") ttk.Combobox(row, textvariable=self.var_array, values=["ULA", "NULA"], state="readonly", width=12).pack(side="left") # ---------- 功能卡 ---------- card2 = ttk.LabelFrame(left, text="功能选择") card2.pack(fill="x", pady=(0, 8)) ttk.Radiobutton(card2, text="波束方向图显示", variable=self.var_function, value="beam").pack(anchor="w") ttk.Radiobutton(card2, text="测角 RMSE 曲线", variable=self.var_function, value="rmse").pack(anchor="w") # ---------- 算法卡 ---------- card3 = ttk.LabelFrame(left, text="算法选择（可多选）") card3.pack(fill="x", pady=(0, 8)) self.alg_vars = { "Analog": tk.BooleanVar(value=True), "Digital": tk.BooleanVar(value=True), "DFT": tk.BooleanVar(value=True), "OMP": tk.BooleanVar(value=True), "TSC-OMP": tk.BooleanVar(value=True), } labels = { "Analog": "纯模拟", "Digital": "纯数字(MVDR)", "DFT": "DFT", "OMP": "传统OMP", "TSC-OMP": "TSC-OMP", } for k in ["Analog", "Digital", "DFT", "OMP", "TSC-OMP"]: ttk.Checkbutton(card3, text=labels[k], variable=self.alg_vars[k]).pack(anchor="w") # ---------- 按钮 ---------- btns = ttk.Frame(left) btns.pack(fill="x", pady=(4, 8)) ttk.Button(btns, text="运行", style="Accent.TButton", command=self.run_main_task).pack(fill="x", pady=3) ttk.Button(btns, text="保存当前图", command=lambda: self._save_panel(self.main_plot)).pack(fill="x", pady=3) # ---------- 说明 ---------- note = ttk.LabelFrame(left, text="说明") note.pack(fill="x", pady=(0, 8)) ttk.Label( note, text="• 设置好您所需的参数，算法可以多选\n" "• 选择您所需要的图表，有波束方向图和测角RMSE比较图可选\n" "• 单击“运行”即可输出所需图表，可以随时保存", justify="left" ).pack(anchor="w", padx=4, pady=4) # ---------- 绘图区域 ---------- self.main_plot = PlotPanel(right) self.main_plot.pack(fill="both", expand=True) # ---------- 状态栏 ---------- status = ttk.Label(self, textvariable=self.status_var, anchor="w") status.pack(fill="x", padx=10, pady=(0, 8)) def _add_labeled_entry(self, master, label, var, width=12): row = ttk.Frame(master) row.pack(fill="x", pady=4) ttk.Label(row, text=label, width=12).pack(side="left") ttk.Entry(row, textvariable=var, width=width).pack(side="left") def _save_panel(self, panel): path = panel.save_figure() if path: self.status_var.set(f"已保存图像: {path}") def get_selected_methods(self): return [k for k, v in self.alg_vars.items() if v.get()] def _set_busy(self, msg): self.config(cursor="watch") self.status_var.set(msg) self.update_idletasks() def _set_idle(self, msg="就绪"): self.config(cursor="") self.status_var.set(msg) self.update_idletasks() def run_main_task(self): try: target = float(self.var_target.get()) clutter = float(self.var_clutter.get()) array_type = self.var_array.get() func = self.var_function.get() window = float(self.var_window.get()) mc = int(self.var_mc.get()) snapshots = int(self.var_snap.get()) methods = self.get_selected_methods() if not methods: messagebox.showwarning("提示", "请至少选择一种算法。") return t0 = time.time() self._set_busy("正在运行，请稍候...") fig = self.main_plot.figure fig.clear() ax = fig.add_subplot(111) if func == "beam": data = self.engine.beam_patterns(target, clutter, array_type, methods, window) theta = data["theta_scan"] name_map = { "Analog": "纯模拟", "Digital": "纯数字(MVDR)", "DFT": "DFT", "OMP": "传统OMP", "TSC-OMP": "TSC-OMP", } ls_map = { "Analog": "-.", "Digital": "-", "DFT": "--", "OMP": ":", "TSC-OMP": "-", } for m in methods: ax.plot( theta, data[m], linestyle=ls_map[m], color=METHOD_COLORS[m], linewidth=2.4, label=name_map[m] ) ax.axvline(target, color="g", linestyle="--", linewidth=1.8, label=f"目标 ({target:.2f}°)") ax.axvline(clutter, color="r", linestyle="--", linewidth=1.8, label=f"干扰 ({clutter:.2f}°)") ax.set_xlim([-60, 60]) ax.set_ylim([-60, 2]) ax.set_title(f"波束方向图 ({array_type})", fontsize=13, fontweight="bold") ax.set_xlabel("角度 θ (°)") ax.set_ylabel("归一化增益 (dB)") ax.grid(True, ls="--", alpha=0.6) ax.legend(loc="lower center", ncol=min(4, len(methods) + 2), fontsize=9) else: curves = self.engine.rmse_curves( target_angle=target, clutter_angle=clutter, array_type=array_type, methods=methods, window_half_width=window, mc=mc, snapshots=snapshots, ) name_map = { "Analog": "纯模拟", "Digital": "纯数字(MVDR)", "DFT": "DFT", "OMP": "传统OMP", "TSC-OMP": "TSC-OMP", } marker_map = { "Analog": "v", "Digital": "o", "DFT": "d", "OMP": "*", "TSC-OMP": "s", } for m in methods: ax.semilogy( curves["snr"], curves[m], marker=marker_map[m], color=METHOD_COLORS[m], linewidth=2.2, label=name_map[m] ) ax.set_title(f"测角 RMSE-SNR 曲线 ({array_type})", fontsize=13, fontweight="bold") ax.set_xlabel("输入 SNR (dB)") ax.set_ylabel("RMSE (°)") ax.set_ylim([1e-3, 20]) ax.grid(True, which="both", ls="--", alpha=0.6) ax.legend() fig.tight_layout() self.main_plot.redraw() self._set_idle(f"运行完成，用时 {time.time() - t0:.2f}s | 设备: {device}") except Exception as e: self._set_idle("运行失败") messagebox.showerror("错误", str(e)) if __name__ == "__main__": app = HBFWorkbench() app.mainloop()