Electromagnetic State Spectrum Semanticization Project
电磁频谱智能分析与语义化编码系统
本项目实现《频谱语义化表征及频谱恢复》的完整工程体系,面向 30–2500 MHz 频段,提供从信号合成、频谱拼接到语义编码的端到端解决方案。
- 频率范围: 30–2500 MHz(可配置)
- 频率分辨率: 1.0 MHz(2471 个频点)
- 干扰类型: 6 种(噪声调频、单音、多音、梳状谱、部分带宽噪声、扫频)
- 输出格式: NPZ 功率谱文件 + PNG 可视化
- 拼接窗口: 13 个 200 MHz 窗口(±100 MHz 半带宽)
- 窗口中心: 130, 330, 400, 600, 800, 1000, 1200, 1400, 1600, 1800, 2000, 2200, 2400 MHz
- 拼接模式: MAX(最大值)/ MEAN(平均值)/ FIRST(优先第一分段)
- 数据来源: IQ
.bin文件(int16/float32)、NPZ 频谱文件
- v1 格式:
SemanticParams- 单区域/多区域自动识别 - v2 格式:
SemanticEncodingV2- 标准化多干扰区域编码 - 恢复精度: 典型误差 < 3 dB(区域内)
- 应用场景: 频谱压缩、传输、存储、对抗推演
采用 4 层模块化架构,清晰分离核心逻辑、算法实现、业务流程和用户接口:
📦 electromagneticState/
├── 📂 src/ # 源码目录
│ └── 📁 electromagnetic_state/ # 唯一顶层包
│ ├── 📁 core/ # Layer 0: 核心数据结构与配置
│ ├── 📁 io/ # Layer 1: 数据输入输出
│ ├── 📁 signal/ # Layer 1: 信号处理算法
│ ├── 📁 semantics/ # Layer 1: 语义编码与解码
│ ├── 📁 pipeline/ # Layer 2: 业务流程编排
│ ├── 📁 visualization/ # 可视化工具
│ └── 📁 viz/ # 轻量绘图工具
├── 📂 scripts/ # Layer 3: 命令行工具
├── 📂 tests/ # 单元测试与集成测试
├── 📂 docs/ # 活跃技术文档
├── 📂 archive/ # 过时文档归档
├── 📂 data/ # 数据输出目录
├── 📂 data_semantic/ # 语义参数配置
├── 📂 evidence/ # 改进建议与评审记录
├── README.md # 项目说明
└── pyproject.toml # Python 项目配置
- Layer 0 (Core): 数据结构与常量,无业务逻辑
- Layer 1 (Processing): 可复用的算法模块,单一职责
- Layer 2 (Pipeline): 组合 Layer 1 完成业务流程
- Layer 3 (CLI): 用户接口,调用 Pipeline 提供服务
依赖方向: Layer 3 → Layer 2 → Layer 1 → Layer 0(严格单向)
# 推荐使用 conda 管理环境
conda create -n electromagnetic-state python=3.11
conda activate electromagnetic-state
# 安装依赖
pip install -r requirements.txt
# (可选)安装开发工具
pip install -r requirements-dev.txtPython 版本: 3.11+ (已在 Python 3.12.9 测试通过)
# 运行全部测试(34 个测试用例)
pytest -q
# 只运行核心功能测试
pytest tests/test_semantics_v2.py tests/test_io_reader.py -vpython scripts/spectrum_cli.py选择任务一、二、三,按提示操作即可生成频谱。
# 任务一:合成干扰频谱
python scripts/spectrum_batch.py compose \
--jammer single_tone:500:25 \
--jammer sweep:1500:20 \
-o data/composed.npz \
--plot data/composed.png
# 任务二:拼接真实 IQ 数据(需先准备 .bin 文件)
python scripts/spectrum_batch.py stitch \
--input-dir data_segment \
--pattern "*.bin" \
--dtype int16 \
--mode max \
-o data/stitched.npz \
--plot data/stitched.png
# 任务三:从语义参数恢复频谱
python scripts/spectrum_batch.py decode-v2 \
--input data_semantic/semantic_case01.json \
-o data/recovered.npzfrom electromagnetic_state.signal.spectrum_composer import SpectrumComposerConfig, add_jammer, compose_spectrum
import numpy as np
# 创建配置
cfg = SpectrumComposerConfig(
freq_min_mhz=30.0,
freq_max_mhz=2500.0,
resolution_mhz=1.0,
noise_floor_db=-100.0,
)
# 添加干扰
add_jammer(cfg, "single_tone", 500.0, 25.0)
add_jammer(cfg, "sweep", 1500.0, 20.0)
# 生成频谱
rng = np.random.default_rng(42)
freq_mhz, power_db = compose_spectrum(cfg, rng=rng)
# 保存结果
np.savez("data/my_spectrum.npz", freq_mhz=freq_mhz, power_db=power_db)
print(f"✓ 生成 {len(freq_mhz)} 个频点的频谱")本工程提供两类入口:
- 顶层 CLI(推荐给使用者):
spectrum_cli.py– 交互式 CLI(任务一/二/三);spectrum_batch.py– 批处理 CLI(compose / stitch / decode / decode‑v2)。
- Python 模块 / pipeline(推荐给开发者与集成方):
src/pipeline/*.py– 可通过python -m electromagnetic_state.pipeline.xxx调用;- 直接导入
electromagnetic_state.signal/electromagnetic_state.semantics/electromagnetic_state.io下的函数。
下面分 CLI 与 Pipeline 两种方式给出典型用法。
python scripts/spectrum_cli.py主菜单:
- 任务一:干扰功率谱合成(配置频段、底噪、干扰类型/Fc/JNR,生成并保存 npz + PNG);
- 任务二:频谱分段拼接(加载多个频谱 npz 或分段文件,选择拼接模式,生成宽带谱);
- 任务三:语义参数频谱恢复(从 JSON 加载语义参数,恢复功率谱并与参考谱对比)。
详细交互步骤、字段说明请见
scripts/spectrum_cli.py代码或归档文档archive/cli_docs/USAGE_CLI.md。
# 组合两个干扰,生成宽带频谱并保存 npz
python scripts/spectrum_batch.py compose \
--jammer single_tone:500:20 \
--jammer sweep:1200:18 \
-o data/composed.npz \
--plot data/composed.pngpython scripts/spectrum_batch.py stitch \
--input-dir data_segment \
--pattern "*.bin" \
--dtype int16 \
--mode max \
--segment-fft-size 512 \
--time-agg-mode mean \
-o data/stitched_30_2500.npz \
--plot data/stitched_30_2500.png其中:
input-dir指向包含多个 200 MHz 分段.bin文件的目录;- 文件名支持两类命名:
- 完整命名:
single_130MHz_204.8MHz_xxx.bin(可从文件名自动推断中心频率与采样率/带宽); - 简化命名:
130MHz.bin(仅中心频率,采样率可用--sample-rate指定;默认 204.8e6 Hz)。
- 完整命名:
segment-fft-size为分段 FFT 的 NFFT(推荐用于大样本.bin,避免“超大 FFT + 绘图卡顿”);设置为0可禁用分段 FFT 并退回单次 FFT(使用--fft-size)。
v1(兼容旧格式,单区域或多区域自动判断):
python scripts/spectrum_batch.py decode \
--input data/demo_semantic_v1.json \
-o data/recovered_v1.npz内部会使用 decode_semantic_auto:
- 若
pos_edge为空/单个 → 使用 v1decode_semantic; - 若
pos_edge有多个 → 使用decode_semantic_multi_region支持多区域。
v2(推荐的多区域语义编码格式):
python scripts/spectrum_batch.py decode-v2 \
--input data_semantic/semantic_case01.json \
-o data/recovered_v2_case01.npzv2 JSON 结构参见 docs/semantic_encoding_requirements.md:
freq_min_mhz/freq_max_mhz/num_bins/noise_floor_db/jammer_regions。
以下示例假定你在工程根目录,且环境中已能导入 electromagnetic_state.*。
from electromagnetic_state.signal.spectrum_composer import SpectrumComposerConfig, add_jammer, compose_spectrum
import numpy as np
cfg = SpectrumComposerConfig(
freq_min_mhz=30.0,
freq_max_mhz=2500.0,
resolution_mhz=1.0,
noise_floor_db=-120.0,
)
add_jammer(cfg, "single_tone", 500.0, 20.0)
add_jammer(cfg, "sweep", 1800.0, 22.0)
freq_mhz, power_db = compose_spectrum(cfg, rng=np.random.default_rng(42))
np.savez("data/task1_composed.npz", freq_mhz=freq_mhz, power_db=power_db)或使用 pipeline CLI:
python -m electromagnetic_state.pipeline.compose_spectrum \
--jammer single_tone:500:20 \
--jammer sweep:1800:22 \
--output-npz data/task1_composed.npz \
--output-png data/task1_composed.pngfrom pathlib import Path
from electromagnetic_state.io.reader import BinDataType
from electromagnetic_state.signal.stitcher import StitchMode
from electromagnetic_state.pipeline.stitch.stitch_real_data import stitch_from_bin_directory
stitched, segments = stitch_from_bin_directory(
directory=Path("data_segment"),
pattern="*.bin",
bin_dtype=BinDataType.INT16,
mode=StitchMode.MAX,
# 推荐:对大样本 .bin 使用分段 FFT,并固定 NFFT(例如 512)
segment_fft_size=512,
time_agg_mode="mean",
)
np.savez(
"data/task2_stitched.npz",
freq_mhz=stitched.freq_mhz,
power_db=stitched.power_db,
coverage_map=stitched.coverage_map,
)import numpy as np
from electromagnetic_state.core.schemas import SemanticParams
from electromagnetic_state.semantics.decode import decode_semantic
params = SemanticParams(
yonghu=1,
youwu=1,
menxian=-120.0,
pos_edge=[500],
neg_edge=[700],
start=470,
end=1470,
fenbianlv=2471,
sinr=np.array([15.0]),
freq_min_mhz=30.0,
freq_max_mhz=2500.0,
)
power_db = decode_semantic(params)评估(v1 自动模式):
python -m electromagnetic_state.pipeline.semantic_eval \
--reference data/stitched_30_2500.npz \
--semantic data/demo_semantic_v1.json \
--report data/semantic_eval_v1.jsonimport numpy as np
from electromagnetic_state.core.schemas import SemanticEncodingV2, JammerRegionV2
from electromagnetic_state.semantics.decode_v2 import decode_semantic_v2
params_v2 = SemanticEncodingV2(
freq_min_mhz=30.0,
freq_max_mhz=2500.0,
num_bins=2471,
noise_floor_db=-80.0,
jammer_regions=[
JammerRegionV2(start_bin=50, end_bin=90, jnr_db=25.0),
JammerRegionV2(start_bin=470, end_bin=490, jnr_db=20.0),
],
)
power_db_v2 = decode_semantic_v2(params_v2)评估(v2):
python -m electromagnetic_state.pipeline.semantic_eval_v2 \
--reference data/stitched_30_2500.npz \
--semantic data_semantic/semantic_case01.json \
--report data/semantic_eval_v2_case01.json- Bug 修复记录 - 详细调试过程与修复方案
- 文档准确性审计 - 文档错误识别与修正报告
- 旧版 CLI 使用指南 - Python/Pipeline 详细使用方法(已过时)
- 旧版 CLI 文档 - 历史 CLI 工具文档
| 参数 | 值 | 说明 |
|---|---|---|
DEFAULT_SEMANTIC_FREQ_MIN_MHZ |
30.0 | 最小频率 |
DEFAULT_SEMANTIC_FREQ_MAX_MHZ |
2500.0 | 最大频率 |
DEFAULT_SEMANTIC_NUM_BINS |
2471 | 频点数量 |
DEFAULT_SEMANTIC_NOISE_FLOOR_DB |
-60.0 | 默认噪声底 |
| 频率分辨率 | 1.0 MHz | (2500-30)/(2471-1) |
DEFAULT_WINDOW_CENTERS_MHZ = (
130, 330, 400, 600, 800, 1000, 1200,
1400, 1600, 1800, 2000, 2200, 2400
)每个窗口半带宽: ±100 MHz
- noise_fm - 噪声调频干扰
- single_tone - 单音干扰
- multi_tone - 多音干扰(3-20 个音调)
- comb - 梳状谱干扰
- partial_band_noise - 部分带宽噪声干扰
- sweep - 扫频干扰
- [2025-11-26] 文档归档整理 - 清理项目结构,归档过时文档
- 创建
archive/归档目录结构(reports, cli_docs) - 归档 14 个历史文档到分类目录
- 根目录保持整洁,仅保留 README.md
- 更新所有文档链接指向正确位置
- 创建
- [2025-11] 项目架构重构 - 完成 4 层模块化架构迁移
- Layer 0:
core/(schemas, config) - Layer 1:
io/,signal/,semantics/ - Layer 2:
pipeline/(compose, stitch, semantic, utils) - Layer 3:
scripts/(spectrum_cli.py, spectrum_batch.py)
- Layer 0:
- [2025-11] 全局变量清理 - 删除 10 个未使用的手动配置文件
- 保留 10 个程序生成的语义测试用例 (
semantic_case01.json~semantic_case10.json)
- 保留 10 个程序生成的语义测试用例 (
- [2025-11] Bug 修复与文档准确性审计 - 完成所有测试通过和文档校正
- 修复噪声波动参数 (commit
7fc78d9) - 完成 14 处文档错误识别与修正
- 生成详细的审计报告和修复记录
- 修复噪声波动参数 (commit
- 总测试数: 34 个
- 通过率: 100% ✅ 所有测试通过
- 最近修复: 噪声波动参数优化 (commit
7fc78d9) -test_high_frequency_jammer现已通过 - 代码质量: 4 层架构清晰,无已知 Bug
- 主分支:
master - 最新提交:
6a8f738- docs: 归档过时文档到 archive/ 目录 - 远程仓库: GitHub
- 状态: ✅ 开发完成,架构重构已合并,所有测试通过
CLI (Layer 3)
↓
Pipeline (Layer 2)
↓
Signal/Semantics/IO (Layer 1)
↓
Core (Layer 0)
频谱合成:
from electromagnetic_state.signal.spectrum_composer import compose_spectrum, add_jammer频谱拼接:
from electromagnetic_state.signal.stitcher import stitch_segments
from electromagnetic_state.pipeline.stitch.stitch_real_data import stitch_from_bin_directory语义解码:
from electromagnetic_state.semantics.decode import decode_semantic # v1 单区域
from electromagnetic_state.semantics.decode_multi import decode_semantic_auto # v1 自动模式
from electromagnetic_state.semantics.decode_v2 import decode_semantic_v2 # v2 标准格式数据 IO:
from electromagnetic_state.io.reader import load_iq_file, load_bin_segments新增干扰类型:
- 在
src/signal/jammers.py添加生成函数xxx_jammer() - 注册到
JAMMER_REGISTRY字典 - 更新 CLI 参数解析器
新增语义编码策略:
- 在
src/semantics/实现编码/解码函数 - 在
src/pipeline/semantic/创建评估流程 - 添加对应的单元测试
v1 格式 (SemanticParams):
- ✅ 兼容现有数据
- ✅ 支持单区域/多区域自动识别
⚠️ 结构相对复杂
v2 格式 (SemanticEncodingV2) - 推荐:
- ✅ 标准化多干扰区域编码
- ✅ 结构清晰,易于扩展
- ✅ 10 个测试用例参考 (
data_semantic/semantic_case*.json)
| 模式 | 适用场景 | 优势 | 劣势 |
|---|---|---|---|
| MAX | 干扰检测、峰值分析 | 保留所有干扰峰值 | 可能放大噪声 |
| MEAN | 噪声抑制、平均功率 | 降低随机噪声影响 | 可能掩盖瞬时干扰 |
| FIRST | 优先级分段处理 | 简单快速 | 忽略后续分段信息 |
这是基于项目需求和实际硬件能力设定的:
- 30 MHz: 短波上限,避开高频噪声
- 2500 MHz: 覆盖主要通信频段(包括 L/S 波段)
- 可自定义: 通过配置参数调整
freq_min_mhz和freq_max_mhz
- 准确的边缘检测: 确保
pos_edge和neg_edge索引精确 - 合理的 JNR 值: 通常设置为 15-30 dB
- 使用 v2 格式: 更清晰的多区域定义
- 验证频点数:
num_bins必须与原始频谱一致 (2471)
# 1. 检查 Python 版本
python --version # 需要 3.11+
# 2. 运行详细测试
pytest -v --tb=short
# 3. 只测试核心功能
pytest tests/test_semantics_v2.py -v
# 4. 查看失败详情
pytest tests/test_semantics_v2.py -vv --tb=long批量拼接示例:
# 方法一:使用 CLI
python scripts/spectrum_batch.py stitch \
--input-dir data_segment \
--pattern "*.bin" \
--dtype int16 \
--mode max \
-o data/batch_stitched.npz
# 方法二:Python 脚本
from electromagnetic_state.io.reader import load_bin_segments, BinDataType
segments = load_bin_segments("data_segment", "*.bin", BinDataType.INT16)
print(f"加载了 {len(segments)} 个分段")data_semantic/semantic_case01.json~semantic_case10.json- 10 个语义编码测试用例data/cli_results/jammer_config.json- 干扰源配置示例
如遇到问题或有改进建议,请:
- 查阅 详细文档 章节
- 检查 常见问题 章节
- 查看项目 GitHub Issues 或提交新问题
本项目仅供学习和研究使用。
🔬 Electromagnetic State Spectrum Semanticization Project
电磁频谱智能分析 · 语义化编码 · 高效恢复