david1465833828/Graduation-Design
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

上传文件至「Graduation Design」

Browse Source
This commit is contained in:
david1465833828
2026-04-30 15:22:19 +08:00
parent 29cbf3b278
commit 6e52c8c714
3 changed files with 1528 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Graduation Design/Massive_Mimo_Beamforming_Testing_Platform.py
+625
View File
@@ -0,0 +1,625 @@
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()
Graduation Design/robust_tsc_omp_cuda_showall.py
+402
View File
@@ -0,0 +1,402 @@
import os
import math
import time
import numpy as np
import torch
import matplotlib.pyplot as plt
# ============================================================
# 0. 全局设置
# ============================================================
plt.rcParams['font.sans-serif'] = [
'Microsoft YaHei', 'SimHei', 'PingFang SC',
'Noto Sans CJK SC', 'Arial Unicode MS', 'DejaVu Sans'
]
plt.rcParams['axes.unicode_minus'] = False
plt.rcParams['mathtext.fontset'] = 'stix'
SAVE_DIR = "robust_tsc_omp_cuda_results"
os.makedirs(SAVE_DIR, exist_ok=True)
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
torch.manual_seed(2024)
np.random.seed(2024)
# ============================================================
# 1. 参数区
# 说明：
# - 本脚本默认使用 CUDA（若可用）
# - 为避免“卡死”，默认采用较稳妥的 Monte Carlo 与扫描步长
# - 所有图都会生成并在脚本结束时统一以窗口形式显示
# ============================================================
ARRAY_TYPE = 'NULA' # 'ULA' 或 'NULA'
N_ant = 24
N_RF = 4
d_lambda = 0.5
target_angle = 5.0
clutter_angle = -15.0
noise_power = 1.0
INR_dB = 30.0
INR_linear = 10 ** (INR_dB / 10)
L_snapshots = 100
Monte_Carlo = 300
SNR_dB_range = np.arange(-10, 22, 2)
theta_prior = 5.0
window_half_width = 5.0
sector_min = theta_prior - window_half_width
sector_max = theta_prior + window_half_width
# TSC-OMP 字典与测角搜索栅格
tsc_grid_np = np.arange(sector_min, sector_max + 0.1, 0.1)
scan_grid_np = np.arange(sector_min, sector_max + 0.001, 0.05) # 比 0.01 更快
# 对角加载系数
alpha_list = [0.00, 0.01, 0.03, 0.05, 0.10]
# 是否保存图片
SAVE_FIG = True
# ============================================================
# 2. 阵列位置
# ============================================================
def get_array_positions(array_type='ULA', N=24, d=0.5):
if array_type.upper() == 'ULA':
pos = np.arange(-(N - 1) / 2, (N - 1) / 2 + 1) * d
return pos.astype(np.float32)
elif array_type.upper() == 'NULA':
# 与论文一致的示意性 NULA
base = 0.7
jump = 1.03
pos = []
x = 0.0
for _ in range(12):
pos.append(x)
x += base
x += (jump - base)
for _ in range(12, 24):
pos.append(x)
x += base
pos = np.array(pos, dtype=np.float32)
pos = pos - np.mean(pos)
return pos
else:
raise ValueError("ARRAY_TYPE 只能是 'ULA' 或 'NULA'")
array_pos_np = get_array_positions(ARRAY_TYPE, N_ant, d_lambda)
array_pos_t = torch.tensor(array_pos_np, dtype=torch.float32, device=device).view(-1, 1)
# ============================================================
# 3. 基础函数（CUDA）
# ============================================================
def sync():
if device.type == 'cuda':
torch.cuda.synchronize()
def steering_vector(theta_deg):
theta = torch.as_tensor(theta_deg, dtype=torch.float32, device=device)
theta_rad = torch.deg2rad(theta)
phase = 2 * torch.pi * array_pos_t * torch.sin(theta_rad)
return torch.exp(1j * phase)
def steering_matrix(theta_deg_array):
theta = torch.as_tensor(theta_deg_array, dtype=torch.float32, device=device).view(1, -1)
theta_rad = torch.deg2rad(theta)
phase = 2 * torch.pi * array_pos_t @ torch.sin(theta_rad)
return torch.exp(1j * phase)
def build_dictionary(angle_grid_np, normalize=True):
A = steering_matrix(angle_grid_np)
if normalize:
A = A / math.sqrt(A.shape[0])
return A
def build_prior_broad(sector_min, sector_max, step=0.5):
angles = np.arange(sector_min, sector_max + 1e-12, step, dtype=np.float32)
A = steering_matrix(angles)
a = torch.sum(A, dim=1, keepdim=True)
a = a / torch.linalg.norm(a)
return a
def extract_omp(w_opt, A_dic, angle_grid_np, N_RF):
"""
CUDA 版 OMP / TSC-OMP
"""
res = w_opt.clone()
F_RF = None
chosen_angles = []
for _ in range(N_RF):
proj = torch.abs(A_dic.mH @ res).flatten()
idx = int(torch.argmax(proj).item())
atom = A_dic[:, idx:idx+1]
F_RF = atom if F_RF is None else torch.cat((F_RF, atom), dim=1)
chosen_angles.append(float(angle_grid_np[idx]))
res = w_opt - F_RF @ (torch.linalg.pinv(F_RF) @ w_opt)
return F_RF, chosen_angles
def build_prior_weight(target_angle, clutter_angle, sector_min, sector_max):
a_clutter = steering_vector(clutter_angle)
R_interf = noise_power * torch.eye(N_ant, dtype=torch.complex64, device=device) \
+ INR_linear * (a_clutter @ a_clutter.mH)
a_prior = build_prior_broad(sector_min, sector_max, step=0.5)
w_prior = torch.linalg.solve(R_interf, a_prior)
w_prior = w_prior / torch.linalg.norm(w_prior)
return w_prior
def calc_hybrid_weight_loaded(F_RF, R_in, a_target, alpha_dl=0.05):
a_eff = F_RF.mH @ a_target
R_eff = F_RF.mH @ R_in @ F_RF
gamma = alpha_dl * torch.real(torch.trace(R_eff)) / R_eff.shape[0]
R_eff_loaded = R_eff + gamma * torch.eye(R_eff.shape[0], dtype=torch.complex64, device=device)
inv_R = torch.linalg.inv(R_eff_loaded)
F_BB = (inv_R @ a_eff) / (a_eff.mH @ inv_R @ a_eff)
w = F_RF @ F_BB
w = w / torch.linalg.norm(w)
cond_number = torch.linalg.cond(R_eff_loaded).real.item()
return w, R_eff_loaded, cond_number
def robust_capon_spectrum_batch(F_RF, X_batch, A_scan, alpha_dl=0.05):
"""
X_batch: [MC, N_ant, L]
A_scan: [N_ant, Ns]
返回：
P: [MC, Ns]
mean_cond: 平均条件数
"""
Y = torch.matmul(F_RF.mH.unsqueeze(0), X_batch) # [MC, N_RF, L]
R_bb = torch.matmul(Y, Y.mH) / X_batch.shape[-1] # [MC, N_RF, N_RF]
trace_R = torch.real(torch.diagonal(R_bb, dim1=-2, dim2=-1).sum(-1))
gamma = alpha_dl * trace_R / F_RF.shape[1]
eye = torch.eye(F_RF.shape[1], dtype=torch.complex64, device=device).unsqueeze(0)
R_bb_loaded = R_bb + gamma[:, None, None] * eye
inv_R = torch.linalg.inv(R_bb_loaded) # [MC, N_RF, N_RF]
A_eff = F_RF.mH @ A_scan # [N_RF, Ns]
# P = 1 / |a^H invR a|
denom = torch.einsum('rn,brs,sn->bn', A_eff.conj(), inv_R, A_eff)
P = 1.0 / torch.clamp(torch.abs(denom), min=1e-12)
cond_vals = torch.linalg.cond(R_bb_loaded).real
mean_cond = torch.mean(cond_vals).item()
return P, mean_cond
def calc_sinr(w, a_target, a_clutter, sig_p):
R_interf = noise_power * torch.eye(N_ant, dtype=torch.complex64, device=device) \
+ INR_linear * (a_clutter @ a_clutter.mH)
num = sig_p * (torch.abs(w.mH @ a_target).item() ** 2)
den = torch.real(w.mH @ R_interf @ w).item()
return 10 * np.log10(max(num / max(den, 1e-12), 1e-12))
# ============================================================
# 4. 固定 TSC-OMP 模拟矩阵
# ============================================================
print(f"当前设备: {device}")
sync()
tic = time.perf_counter()
w_prior = build_prior_weight(
target_angle=target_angle,
clutter_angle=clutter_angle,
sector_min=sector_min,
sector_max=sector_max
)
A_dic_tsc = build_dictionary(tsc_grid_np, normalize=True)
F_RF_tsc, chosen_angles = extract_omp(w_prior, A_dic_tsc, tsc_grid_np, N_RF)
a_target = steering_vector(target_angle)
a_clutter = steering_vector(clutter_angle)
sync()
print(f"TSC-OMP 选中的模拟原子角度: {chosen_angles}")
print(f"固定模拟矩阵构建耗时: {time.perf_counter() - tic:.3f} s")
# ============================================================
# 5. 静态方向图：不同加载系数
# ============================================================
sig_power_static = 10 ** (0 / 10.0)
s_t = np.sqrt(sig_power_static / 2) * (
torch.randn(1, L_snapshots, dtype=torch.float32, device=device)
+ 1j * torch.randn(1, L_snapshots, dtype=torch.float32, device=device)
)
s_c = np.sqrt(INR_linear / 2) * (
torch.randn(1, L_snapshots, dtype=torch.float32, device=device)
+ 1j * torch.randn(1, L_snapshots, dtype=torch.float32, device=device)
)
n_t = np.sqrt(noise_power / 2) * (
torch.randn(N_ant, L_snapshots, dtype=torch.float32, device=device)
+ 1j * torch.randn(N_ant, L_snapshots, dtype=torch.float32, device=device)
)
X_static = a_target @ s_t + a_clutter @ s_c + n_t
R_in_static = (X_static @ X_static.mH) / L_snapshots
theta_scan_np = np.arange(-60, 60.1, 0.1)
A_bp = steering_matrix(theta_scan_np)
plt.figure(figsize=(10, 6))
for alpha_dl in alpha_list:
w_hyb, _, _ = calc_hybrid_weight_loaded(F_RF_tsc, R_in_static, a_target, alpha_dl=alpha_dl)
BP = torch.abs(w_hyb.mH @ A_bp).flatten() ** 2
BP = 10 * torch.log10(BP / torch.max(BP) + 1e-12)
plt.plot(theta_scan_np, BP.detach().cpu().numpy(), linewidth=2, label=f'α={alpha_dl:.2f}')
plt.axvline(target_angle, color='g', linestyle='--', label=f'目标 {target_angle:.2f}°')
plt.axvline(clutter_angle, color='r', linestyle='--', label=f'杂波 {clutter_angle:.2f}°')
plt.xlabel('角度 θ (°)')
plt.ylabel('归一化增益 (dB)')
plt.title(f'不同对角加载系数下 TSC-OMP 波束方向图 ({ARRAY_TYPE})')
plt.grid(True, linestyle='--', alpha=0.5)
plt.legend()
plt.tight_layout()
if SAVE_FIG:
plt.savefig(os.path.join(SAVE_DIR, f'BP_loaded_compare_{ARRAY_TYPE}.png'), dpi=300)
# ============================================================
# 6. 输出 SINR / RMSE / 条件数 对比
# ============================================================
SINR_loaded = {alpha: [] for alpha in alpha_list}
RMSE_loaded = {alpha: [] for alpha in alpha_list}
COND_loaded = {alpha: [] for alpha in alpha_list}
A_scan = steering_matrix(scan_grid_np)
for snr_db in SNR_dB_range:
sync()
tic = time.perf_counter()
sig_p = 10 ** (snr_db / 10.0)
# 一次性批量生成 Monte Carlo 数据
s_t = np.sqrt(sig_p / 2) * (
torch.randn(Monte_Carlo, 1, L_snapshots, dtype=torch.float32, device=device)
+ 1j * torch.randn(Monte_Carlo, 1, L_snapshots, dtype=torch.float32, device=device)
)
s_c = np.sqrt(INR_linear / 2) * (
torch.randn(Monte_Carlo, 1, L_snapshots, dtype=torch.float32, device=device)
+ 1j * torch.randn(Monte_Carlo, 1, L_snapshots, dtype=torch.float32, device=device)
)
n_t = np.sqrt(noise_power / 2) * (
torch.randn(Monte_Carlo, N_ant, L_snapshots, dtype=torch.float32, device=device)
+ 1j * torch.randn(Monte_Carlo, N_ant, L_snapshots, dtype=torch.float32, device=device)
)
X_batch = a_target.unsqueeze(0) * s_t + a_clutter.unsqueeze(0) * s_c + n_t
R_in_batch = torch.matmul(X_batch, X_batch.mH) / L_snapshots
for alpha_dl in alpha_list:
# --- SINR：使用每个 Monte Carlo 样本求权，最后取平均 ---
sinr_vals = []
cond_eff_vals = []
for mc in range(Monte_Carlo):
w_hyb, _, cond_eff = calc_hybrid_weight_loaded(
F_RF_tsc, R_in_batch[mc], a_target, alpha_dl=alpha_dl
)
sinr_vals.append(calc_sinr(w_hyb, a_target, a_clutter, sig_p))
cond_eff_vals.append(cond_eff)
SINR_loaded[alpha_dl].append(float(np.mean(sinr_vals)))
# --- RMSE 与 R_bb 条件数：批量 Capon 谱 ---
P, mean_cond_bb = robust_capon_spectrum_batch(
F_RF_tsc, X_batch, A_scan, alpha_dl=alpha_dl
)
est_idx = torch.argmax(P, dim=1)
est_angles = torch.as_tensor(scan_grid_np, dtype=torch.float32, device=device)[est_idx]
rmse = torch.sqrt(torch.mean((est_angles - target_angle) ** 2)).item()
RMSE_loaded[alpha_dl].append(rmse)
COND_loaded[alpha_dl].append(mean_cond_bb)
sync()
print(f"SNR={snr_db:>3} dB 完成，耗时 {time.perf_counter() - tic:.2f} s")
# ============================================================
# 7. 输出 SINR 图
# ============================================================
plt.figure(figsize=(10, 6))
for alpha_dl in alpha_list:
plt.plot(SNR_dB_range, SINR_loaded[alpha_dl], marker='o', linewidth=2, label=f'α={alpha_dl:.2f}')
plt.xlabel('输入 SNR (dB)')
plt.ylabel('输出 SINR (dB)')
plt.title(f'不同对角加载系数下 TSC-OMP 输出 SINR 对比 ({ARRAY_TYPE})')
plt.grid(True, linestyle='--', alpha=0.5)
plt.legend()
plt.tight_layout()
if SAVE_FIG:
plt.savefig(os.path.join(SAVE_DIR, f'SINR_loaded_compare_{ARRAY_TYPE}.png'), dpi=300)
# ============================================================
# 8. RMSE 图
# ============================================================
plt.figure(figsize=(10, 6))
for alpha_dl in alpha_list:
plt.semilogy(SNR_dB_range, RMSE_loaded[alpha_dl], marker='s', linewidth=2, label=f'α={alpha_dl:.2f}')
plt.xlabel('输入 SNR (dB)')
plt.ylabel('测角 RMSE (°)')
plt.title(f'不同对角加载系数下 TSC-OMP 测角 RMSE 对比 ({ARRAY_TYPE})')
plt.grid(True, which='both', linestyle='--', alpha=0.5)
plt.legend()
plt.tight_layout()
if SAVE_FIG:
plt.savefig(os.path.join(SAVE_DIR, f'RMSE_loaded_compare_{ARRAY_TYPE}.png'), dpi=300)
# ============================================================
# 9. 条件数图
# ============================================================
plt.figure(figsize=(10, 6))
for alpha_dl in alpha_list:
plt.semilogy(SNR_dB_range, COND_loaded[alpha_dl], marker='d', linewidth=2, label=f'α={alpha_dl:.2f}')
plt.xlabel('输入 SNR (dB)')
plt.ylabel('低维协方差矩阵条件数')
plt.title(f'不同对角加载系数下低维协方差矩阵条件数 ({ARRAY_TYPE})')
plt.grid(True, which='both', linestyle='--', alpha=0.5)
plt.legend()
plt.tight_layout()
if SAVE_FIG:
plt.savefig(os.path.join(SAVE_DIR, f'COND_loaded_compare_{ARRAY_TYPE}.png'), dpi=300)
# ============================================================
# 10. 低SNR点 α 扫描汇总图
# ============================================================
low_snr_pick = 0
alpha_arr = np.array(alpha_list, dtype=float)
rmse_pick = np.array([RMSE_loaded[a][np.where(SNR_dB_range == low_snr_pick)[0][0]] for a in alpha_list], dtype=float)
cond_pick = np.array([COND_loaded[a][np.where(SNR_dB_range == low_snr_pick)[0][0]] for a in alpha_list], dtype=float)
fig = plt.figure(figsize=(10, 6))
ax1 = fig.add_subplot(111)
ax1.plot(alpha_arr, rmse_pick, marker='o', linewidth=2, color='#0072BD', label='RMSE')
ax1.set_xlabel('对角加载系数 α')
ax1.set_ylabel('RMSE (°)', color='#0072BD')
ax1.tick_params(axis='y', labelcolor='#0072BD')
ax1.grid(True, linestyle='--', alpha=0.5)
ax2 = ax1.twinx()
ax2.semilogy(alpha_arr, cond_pick, marker='s', linewidth=2, color='#D95319', label='条件数')
ax2.set_ylabel('条件数', color='#D95319')
ax2.tick_params(axis='y', labelcolor='#D95319')
plt.title(f'低SNR={low_snr_pick} dB 下加载系数对 RMSE 与条件数的影响 ({ARRAY_TYPE})')
fig.tight_layout()
if SAVE_FIG:
plt.savefig(os.path.join(SAVE_DIR, f'alpha_sweep_lowSNR_{ARRAY_TYPE}.png'), dpi=300)
# ============================================================
# 11. 打印总结
# ============================================================
print("\n===== 鲁棒性提升实验完成 =====")
for alpha_dl in alpha_list:
print(f"α={alpha_dl:.2f} : "
f"平均SINR={np.mean(SINR_loaded[alpha_dl]):.3f} dB, "
f"平均RMSE={np.mean(RMSE_loaded[alpha_dl]):.4f}°, "
f"平均条件数={np.mean(COND_loaded[alpha_dl]):.3e}")
print(f"\n结果图片已保存到: {SAVE_DIR}")
print("即将以窗口形式显示所有图片……")
plt.show()
Graduation Design/verify_pseudo_peak_omp_tsc.py
+501
View File
@@ -0,0 +1,501 @@
import numpy as np
import matplotlib.pyplot as plt
from matplotlib.patches import Rectangle
import os
# ============================================================
# 0. 全局绘图设置（尽量兼容中文）
# ============================================================
plt.rcParams['font.sans-serif'] = [
'Microsoft YaHei', 'SimHei', 'PingFang SC',
'Noto Sans CJK SC', 'Arial Unicode MS', 'DejaVu Sans'
]
plt.rcParams['axes.unicode_minus'] = False
# ============================================================
# 1. 用户参数区（你可以根据论文需要自行调整）
# ============================================================
ARRAY_TYPE = 'ULA' # 'ULA' 或 'NULA'
N_ant = 24 # 阵元数
N_RF = 4 # 射频链数
d_lambda = 0.5 # 阵元间距（单位：波长），ULA 用
target_angle = 5.0 # 目标角度
clutter_angle = -15.0 # 强杂波角度
sector_min = 0.0 # TSC-OMP 扇区下界
sector_max = 10.0 # TSC-OMP 扇区上界
clutter_band_half = 2.0 # 杂波邻域半宽（用于统计“杂波误选率”）
noise_power = 1.0
SNR_dB = -15.0 # 故意设置得较低，便于显现伪峰
INR_dB = 35.0 # 强干扰
L_snapshots = 48 # 快拍数，故意设置得不高，便于显现伪峰
diag_loading_ratio = 0.01 # 对角加载比例
MC = 400 # Monte Carlo 次数
# 字典设置
full_grid = np.arange(-90.0, 90.0 + 0.5, 0.5) # 传统OMP：全空间字典
tsc_grid = np.arange(sector_min, sector_max + 0.1, 0.1) # TSC-OMP：局部字典
# 宽波束先验
broad_step = 0.5
# 输出文件夹
save_dir = 'pseudo_peak_results'
os.makedirs(save_dir, exist_ok=True)
# ============================================================
# 2. 阵列位置定义
# ============================================================
def get_array_positions(array_type='ULA', N=24, d=0.5):
"""
返回阵列位置（单位：波长）
"""
if array_type.upper() == 'ULA':
pos = np.arange(-(N - 1) / 2, (N - 1) / 2 + 1) * d
return pos.astype(float)
elif array_type.upper() == 'NULA':
# 一个示例性的 NULA 阵列位置（非均匀、递增排序、近似关于0分布）
# 你也可以改成与你论文代码完全一致的 NULA 位置
pos = np.array([
-5.75, -5.10, -4.40, -3.85, -3.15, -2.60,
-2.05, -1.45, -0.95, -0.35, 0.20, 0.75,
1.25, 1.85, 2.45, 3.05, 3.70, 4.30,
4.95, 5.65, 6.35, 7.05, 7.85, 8.70
])
pos = pos - np.mean(pos)
return pos.astype(float)
else:
raise ValueError("ARRAY_TYPE 只能是 'ULA' 或 'NULA'")
array_pos = get_array_positions(ARRAY_TYPE, N_ant, d_lambda)
# ============================================================
# 3. 基本函数
# ============================================================
def steering_vector(theta_deg, array_pos):
"""
阵列导向矢量 a(theta)
theta_deg : 角度（度）
array_pos : 阵元位置（单位：波长）
"""
theta_rad = np.deg2rad(theta_deg)
phase = 2 * np.pi * array_pos * np.sin(theta_rad)
a = np.exp(1j * phase).reshape(-1, 1)
return a
def build_dictionary(angle_grid, array_pos, normalize=True):
"""
构造字典矩阵 A = [a(theta1), a(theta2), ...]
"""
A = np.hstack([steering_vector(ang, array_pos) for ang in angle_grid])
if normalize:
A = A / np.sqrt(A.shape[0])
return A
def omp_trace(w_opt, A_dic, angle_grid, N_RF):
"""
传统 OMP / TSC-OMP 的公共实现：只是在字典上不同
返回：
F_RF : 最终模拟矩阵
chosen_ang : 每一步选中的角度
proj_hist : 每一步投影谱 |A^H r|
"""
res = w_opt.copy()
F_RF = np.zeros((A_dic.shape[0], 0), dtype=np.complex128)
chosen_ang = []
proj_hist = []
for _ in range(N_RF):
proj = np.abs(A_dic.conj().T @ res).flatten()
proj_hist.append(proj.copy())
idx = np.argmax(proj)
chosen_ang.append(float(angle_grid[idx]))
atom = A_dic[:, idx:idx+1]
F_RF = np.hstack((F_RF, atom))
coeff = np.linalg.pinv(F_RF) @ w_opt
res = w_opt - F_RF @ coeff
return F_RF, np.array(chosen_ang), proj_hist
def build_broad_prior_vector(array_pos, sector_min, sector_max, step=0.5):
"""
宽波束先验导向向量：对目标扇区做叠加
"""
a_broad = np.zeros((len(array_pos), 1), dtype=np.complex128)
for th in np.arange(sector_min, sector_max + 1e-12, step):
a_broad += steering_vector(th, array_pos)
a_broad /= np.linalg.norm(a_broad)
return a_broad
def build_sample_prior(array_pos,
target_angle,
clutter_angle,
SNR_dB,
INR_dB,
noise_power,
L_snapshots,
diag_loading_ratio,
sector_min,
sector_max,
broad_step=0.5):
"""
构造有限快拍样本协方差下的先验数字权值 w_prior
这是让传统 OMP 更容易显现伪峰风险的关键
"""
N_ant = len(array_pos)
sig_p = 10 ** (SNR_dB / 10.0)
INR_linear = 10 ** (INR_dB / 10.0)
a_t = steering_vector(target_angle, array_pos)
a_c = steering_vector(clutter_angle, array_pos)
s_t = np.sqrt(sig_p / 2) * (
np.random.randn(1, L_snapshots) + 1j * np.random.randn(1, L_snapshots)
)
s_c = np.sqrt(INR_linear / 2) * (
np.random.randn(1, L_snapshots) + 1j * np.random.randn(1, L_snapshots)
)
n_t = np.sqrt(noise_power / 2) * (
np.random.randn(N_ant, L_snapshots) + 1j * np.random.randn(N_ant, L_snapshots)
)
X = a_t @ s_t + a_c @ s_c + n_t
R_hat = (X @ X.conj().T) / L_snapshots
dl = diag_loading_ratio * np.real(np.trace(R_hat)) / N_ant
R_hat = R_hat + dl * np.eye(N_ant)
a_broad = build_broad_prior_vector(array_pos, sector_min, sector_max, broad_step)
# 用低快拍样本协方差构造“先验数字权值”
w_prior = np.linalg.solve(R_hat, a_broad)
w_prior /= np.linalg.norm(w_prior)
return w_prior
def count_stats(chosen_angles, sector_min, sector_max, clutter_angle, clutter_band_half):
"""
统计：
P_hit : 目标扇区命中率
P_clutter : 杂波误选率
P_out : 扇区外误选率
"""
chosen_angles = np.array(chosen_angles)
hit = np.sum((chosen_angles >= sector_min) & (chosen_angles <= sector_max))
clutter = np.sum((chosen_angles >= clutter_angle - clutter_band_half) &
(chosen_angles <= clutter_angle + clutter_band_half))
out = np.sum((chosen_angles < sector_min) | (chosen_angles > sector_max))
return hit, clutter, out
def compute_badness_score(proj, angle_grid, sector_min, sector_max, clutter_angle, clutter_band_half):
"""
用于自动挑选一个“传统OMP伪峰最明显”的代表性实验样本
"""
proj = np.array(proj).flatten()
target_mask = (angle_grid >= sector_min) & (angle_grid <= sector_max)
clutter_mask = (angle_grid >= clutter_angle - clutter_band_half) & (angle_grid <= clutter_angle + clutter_band_half)
out_mask = ~target_mask
target_peak = np.max(proj[target_mask]) if np.any(target_mask) else 1e-12
clutter_peak = np.max(proj[clutter_mask]) if np.any(clutter_mask) else 0.0
out_peak = np.max(proj[out_mask]) if np.any(out_mask) else 0.0
score = max(clutter_peak / (target_peak + 1e-12),
out_peak / (target_peak + 1e-12))
return score
# ============================================================
# 4. 构造字典
# ============================================================
A_dic_full = build_dictionary(full_grid, array_pos, normalize=True)
A_dic_tsc = build_dictionary(tsc_grid, array_pos, normalize=True)
# ============================================================
# 5. Monte Carlo 验证
# ============================================================
all_angles_full = []
all_angles_tsc = []
full_hit_total = 0
full_clutter_total = 0
full_out_total = 0
tsc_hit_total = 0
tsc_clutter_total = 0
tsc_out_total = 0
# 保存一个“代表性最差样本”，用于画第一步投影谱
best_bad_score = -1.0
rep_full_proj0 = None
rep_tsc_proj0 = None
rep_full_chosen = None
rep_tsc_chosen = None
for mc in range(MC):
w_prior = build_sample_prior(
array_pos=array_pos,
target_angle=target_angle,
clutter_angle=clutter_angle,
SNR_dB=SNR_dB,
INR_dB=INR_dB,
noise_power=noise_power,
L_snapshots=L_snapshots,
diag_loading_ratio=diag_loading_ratio,
sector_min=sector_min,
sector_max=sector_max,
broad_step=broad_step
)
# 传统OMP（全空间字典）
_, chosen_full, proj_hist_full = omp_trace(w_prior, A_dic_full, full_grid, N_RF)
# TSC-OMP（目标扇区局部字典）
_, chosen_tsc, proj_hist_tsc = omp_trace(w_prior, A_dic_tsc, tsc_grid, N_RF)
all_angles_full.extend(chosen_full.tolist())
all_angles_tsc.extend(chosen_tsc.tolist())
h, c, o = count_stats(chosen_full, sector_min, sector_max, clutter_angle, clutter_band_half)
full_hit_total += h
full_clutter_total += c
full_out_total += o
h, c, o = count_stats(chosen_tsc, sector_min, sector_max, clutter_angle, clutter_band_half)
tsc_hit_total += h
tsc_clutter_total += c
tsc_out_total += o
# 选一个“传统OMP最容易看到伪峰”的样本
cur_score = compute_badness_score(proj_hist_full[0], full_grid,
sector_min, sector_max,
clutter_angle, clutter_band_half)
# 优先考虑“传统OMP至少有一个原子落在扇区外或杂波邻域”的样本
bad_event = np.any((chosen_full < sector_min) | (chosen_full > sector_max)) or \
np.any((chosen_full >= clutter_angle - clutter_band_half) &
(chosen_full <= clutter_angle + clutter_band_half))
if bad_event and cur_score > best_bad_score:
best_bad_score = cur_score
rep_full_proj0 = proj_hist_full[0]
rep_tsc_proj0 = proj_hist_tsc[0]
rep_full_chosen = chosen_full.copy()
rep_tsc_chosen = chosen_tsc.copy()
# 如果一个“明确坏样本”都没抓到，就退而选投影谱最坏的那一个
if rep_full_proj0 is None:
# 再跑一遍，抓 score 最高的
best_bad_score = -1.0
for mc in range(MC):
w_prior = build_sample_prior(
array_pos=array_pos,
target_angle=target_angle,
clutter_angle=clutter_angle,
SNR_dB=SNR_dB,
INR_dB=INR_dB,
noise_power=noise_power,
L_snapshots=L_snapshots,
diag_loading_ratio=diag_loading_ratio,
sector_min=sector_min,
sector_max=sector_max,
broad_step=broad_step
)
_, chosen_full, proj_hist_full = omp_trace(w_prior, A_dic_full, full_grid, N_RF)
_, chosen_tsc, proj_hist_tsc = omp_trace(w_prior, A_dic_tsc, tsc_grid, N_RF)
cur_score = compute_badness_score(proj_hist_full[0], full_grid,
sector_min, sector_max,
clutter_angle, clutter_band_half)
if cur_score > best_bad_score:
best_bad_score = cur_score
rep_full_proj0 = proj_hist_full[0]
rep_tsc_proj0 = proj_hist_tsc[0]
rep_full_chosen = chosen_full.copy()
rep_tsc_chosen = chosen_tsc.copy()
# ============================================================
# 6. 统计量计算
# ============================================================
total_selections = MC * N_RF
P_hit_full = full_hit_total / total_selections
P_clutter_full = full_clutter_total / total_selections
P_out_full = full_out_total / total_selections
P_hit_tsc = tsc_hit_total / total_selections
P_clutter_tsc = tsc_clutter_total / total_selections
P_out_tsc = tsc_out_total / total_selections
print("==================================================")
print(f"阵列类型: {ARRAY_TYPE}")
print(f"SNR = {SNR_dB:.1f} dB, INR = {INR_dB:.1f} dB, 快拍数 = {L_snapshots}, MC = {MC}")
print("--------------------------------------------------")
print("传统 OMP（全空间字典）:")
print(f"目标扇区命中率 P_hit = {P_hit_full:.4f}")
print(f"杂波误选率 P_clutter = {P_clutter_full:.4f}")
print(f"扇区外误选率 P_out = {P_out_full:.4f}")
print("--------------------------------------------------")
print("TSC-OMP（目标扇区约束字典）:")
print(f"目标扇区命中率 P_hit = {P_hit_tsc:.4f}")
print(f"杂波误选率 P_clutter = {P_clutter_tsc:.4f}")
print(f"扇区外误选率 P_out = {P_out_tsc:.4f}")
print("==================================================")
# 保存统计结果到 txt
with open(os.path.join(save_dir, 'pseudo_peak_statistics.txt'), 'w', encoding='utf-8') as f:
f.write(f"阵列类型: {ARRAY_TYPE}\n")
f.write(f"SNR = {SNR_dB:.1f} dB, INR = {INR_dB:.1f} dB, 快拍数 = {L_snapshots}, MC = {MC}\n\n")
f.write("传统 OMP（全空间字典）:\n")
f.write(f"P_hit = {P_hit_full:.6f}\n")
f.write(f"P_clutter = {P_clutter_full:.6f}\n")
f.write(f"P_out = {P_out_full:.6f}\n\n")
f.write("TSC-OMP（目标扇区约束字典）:\n")
f.write(f"P_hit = {P_hit_tsc:.6f}\n")
f.write(f"P_clutter = {P_clutter_tsc:.6f}\n")
f.write(f"P_out = {P_out_tsc:.6f}\n")
# ============================================================
# 7. 画图1：第一步投影谱对比（最适合论文说明“伪峰”）
# ============================================================
fig1, axes = plt.subplots(2, 1, figsize=(10, 9))
# 上图：传统OMP第一步投影谱
ax = axes[0]
ax.plot(full_grid, rep_full_proj0, 'b-', linewidth=1.8, label='传统OMP第一步投影谱')
ax.axvline(target_angle, color='g', linestyle='--', linewidth=1.5, label=f'目标 {target_angle:.2f}°')
ax.axvline(clutter_angle, color='r', linestyle='--', linewidth=1.5, label=f'杂波 {clutter_angle:.2f}°')
# 标出目标扇区
ymax = 1.05 * np.max(rep_full_proj0)
rect = Rectangle((sector_min, 0), sector_max - sector_min, ymax,
facecolor='green', alpha=0.10, edgecolor=None)
ax.add_patch(rect)
ax.text((sector_min + sector_max) / 2, 0.93 * ymax, '目标扇区', color='green',
ha='center', va='top', fontsize=11)
# 标出杂波邻域
rect2 = Rectangle((clutter_angle - clutter_band_half, 0), 2 * clutter_band_half, ymax,
facecolor='red', alpha=0.10, edgecolor=None)
ax.add_patch(rect2)
ax.text(clutter_angle, 0.80 * ymax, '杂波邻域', color='red',
ha='center', va='top', fontsize=11)
# 标出传统OMP选中的角度
for idx, ang in enumerate(rep_full_chosen):
val = rep_full_proj0[np.argmin(np.abs(full_grid - ang))]
ax.plot(ang, val, 'ko')
ax.text(ang, val + 0.02 * ymax, f'{idx+1}:{ang:.1f}°', fontsize=9, ha='center')
ax.set_title('传统OMP第一步投影谱', fontsize=13, fontweight='bold')
ax.set_xlabel('角度 (°)')
ax.set_ylabel(r'$|a^H(\theta)r^{(0)}|$')
ax.grid(True, linestyle='--', alpha=0.5)
ax.set_xlim([-90, 90])
ax.legend(loc='upper right', fontsize=10)
# 下图：TSC-OMP第一步投影谱
ax = axes[1]
ax.plot(tsc_grid, rep_tsc_proj0, 'm-', linewidth=1.8, label='TSC-OMP第一步投影谱')
ax.axvline(target_angle, color='g', linestyle='--', linewidth=1.5, label=f'目标 {target_angle:.2f}°')
for idx, ang in enumerate(rep_tsc_chosen):
val = rep_tsc_proj0[np.argmin(np.abs(tsc_grid - ang))]
ax.plot(ang, val, 'ko')
ax.text(ang, val + 0.02 * np.max(rep_tsc_proj0), f'{idx+1}:{ang:.1f}°', fontsize=9, ha='center')
ax.set_title('TSC-OMP第一步投影谱', fontsize=13, fontweight='bold')
ax.set_xlabel('角度 (°)')
ax.set_ylabel(r'$|a^H(\theta)r^{(0)}|$')
ax.grid(True, linestyle='--', alpha=0.5)
ax.set_xlim([sector_min, sector_max])
ax.legend(loc='upper right', fontsize=10)
plt.tight_layout()
fig1.savefig(os.path.join(save_dir, f'Fig1_projection_compare_{ARRAY_TYPE}.png'), dpi=300, bbox_inches='tight')
# ============================================================
# 8. 画图2：被选原子角度分布直方图
# ============================================================
fig2, axes = plt.subplots(2, 1, figsize=(10, 8), sharex=False)
# 传统OMP
ax = axes[0]
bins_full = np.arange(-90, 90 + 1, 1.0)
ax.hist(all_angles_full, bins=bins_full, color='steelblue', edgecolor='black', alpha=0.85)
ax.axvline(target_angle, color='g', linestyle='--', linewidth=1.5, label='目标')
ax.axvline(clutter_angle, color='r', linestyle='--', linewidth=1.5, label='杂波')
ax.axvspan(sector_min, sector_max, color='green', alpha=0.10, label='目标扇区')
ax.set_title('传统OMP所选原子角度分布', fontsize=13, fontweight='bold')
ax.set_xlabel('被选原子角度 (°)')
ax.set_ylabel('出现次数')
ax.grid(True, linestyle='--', alpha=0.5)
ax.legend(fontsize=10)
# TSC-OMP
ax = axes[1]
bins_tsc = np.arange(sector_min, sector_max + 0.2, 0.2)
ax.hist(all_angles_tsc, bins=bins_tsc, color='orchid', edgecolor='black', alpha=0.85)
ax.axvline(target_angle, color='g', linestyle='--', linewidth=1.5, label='目标')
ax.axvspan(sector_min, sector_max, color='green', alpha=0.10, label='目标扇区')
ax.set_title('TSC-OMP所选原子角度分布', fontsize=13, fontweight='bold')
ax.set_xlabel('被选原子角度 (°)')
ax.set_ylabel('出现次数')
ax.grid(True, linestyle='--', alpha=0.5)
ax.legend(fontsize=10)
plt.tight_layout()
fig2.savefig(os.path.join(save_dir, f'Fig2_hist_compare_{ARRAY_TYPE}.png'), dpi=300, bbox_inches='tight')
# ============================================================
# 9. 画图3：伪峰统计指标柱状图
# ============================================================
fig3, ax = plt.subplots(figsize=(9, 6))
metrics = ['目标扇区命中率', '杂波误选率', '扇区外误选率']
omp_values = [P_hit_full, P_clutter_full, P_out_full]
tsc_values = [P_hit_tsc, P_clutter_tsc, P_out_tsc]
x = np.arange(len(metrics))
width = 0.34
bars1 = ax.bar(x - width/2, omp_values, width, label='传统OMP', color='steelblue')
bars2 = ax.bar(x + width/2, tsc_values, width, label='TSC-OMP', color='orchid')
for bars in [bars1, bars2]:
for b in bars:
h = b.get_height()
ax.text(b.get_x() + b.get_width()/2, h + 0.01, f'{h:.3f}',
ha='center', va='bottom', fontsize=10)
ax.set_xticks(x)
ax.set_xticklabels(metrics, fontsize=11)
ax.set_ylim([0, 1.08])
ax.set_ylabel('概率 / 比例', fontsize=12)
ax.set_title('传统OMP与TSC-OMP伪峰统计指标对比', fontsize=13, fontweight='bold')
ax.grid(True, axis='y', linestyle='--', alpha=0.5)
ax.legend(fontsize=11)
plt.tight_layout()
fig3.savefig(os.path.join(save_dir, f'Fig3_stat_compare_{ARRAY_TYPE}.png'), dpi=300, bbox_inches='tight')
# ============================================================
# 10. 显示图像
# ============================================================
plt.show()