盲源分离 (Blind Source Separation) 与波束成形 (Beamforming) 算法的研究仓库。基于 PyTorch,采用 Hydra 配置管理,面向多通道音频场景。
# 环境:使用 conda py312 环境
conda activate py312
# 跑测试
conda run -n py312 python -m pytest test/ -v
# 运行 BSS 算法
python run_bss.py audio.input_file=path/to/audio.wav algorithm=five_online
# 运行波束成形
python run_bf.py audio.input_file=path/to/audio.wav algorithm=mvdr
# Hydra 参数覆盖
python run_bss.py algorithm=ilrma_v2 algorithm.params.n_iter=50 output.output_dir=my_outputssrc/
├── bss/ # 盲源分离核心包
│ ├── base.py # BSSBase(torch.nn.Module, ABC)
│ ├── registry.py # @register_bss 注册器
│ ├── cli.py # Hydra CLI 入口
│ ├── audio.py # STFT 和音频 I/O
│ ├── utils.py # contrast_weights, nmf_update, select_target_index
│ ├── configs/algorithm/ # 算法 YAML 配置(11 个)
│ ├── iva/ # IVA 系列算法(7 个)
│ ├── ilrma/ # ILRMA 系列算法(5 个)
│ └── rcscme/ # RCSCME(1 个)
├── bf/ # 波束成形核心包
│ ├── base.py # BFBase(torch.nn.Module, ABC)
│ ├── registry.py # @register_bf 注册器
│ ├── cli.py # Hydra CLI 入口
│ ├── mvdr.py # MVDR 波束成形器
│ └── simulation.py # RIR 仿真、验证场景
├── spatial.py # 空间处理工具(协方差、导向向量、EVD)
├── audio.py # 兼容层 → src.bss.audio
└── utils.py # 兼容层 → src.spatial
test/ # pytest 测试
scripts/ # 验证脚本(mvdr_validation 等)
tools/ # 设备评估、算法审计脚本(30+)
configs/ # 顶层 Hydra 配置
| 名称 | 路径 | 在线 | 家族 |
|---|---|---|---|
| IVA_NG | bss/iva/iva_ng.py |
✗ | iva |
| AUX_IVA_ISS | bss/iva/aux_iva_iss.py |
✗ | iva |
| AUX_IVA_ISS_ONLINE | bss/iva/aux_iva_iss_online.py |
✓ | iva |
| AUX_OVER_IVA | bss/iva/aux_over_iva.py |
✗ | iva |
| AUX_OVER_IVA_ONLINE | bss/iva/aux_over_iva_online.py |
✓ | iva |
| FIVE | bss/iva/five.py |
✗ | iva |
| FIVE_ONLINE | bss/iva/five_online.py |
✓ | iva |
| ILRMA | bss/ilrma/ilrma.py |
✗ | ilrma |
| ILRMA_V2 | bss/ilrma/ilrma_v2.py |
✗ | ilrma |
| ILRMA_SR | bss/ilrma/ilrma_sr.py |
✗ | ilrma |
| ILRMA_REALTIME | bss/ilrma/ilrma_real_time.py |
✓ | ilrma |
| ILRMA_NOISY | bss/ilrma/ilrma_noisy.py |
✗ | ilrma |
| RCSCME | bss/rcscme/rcscme.py |
✗ | rcscme |
| 名称 | 路径 | 说明 |
|---|---|---|
| MVDR | bf/mvdr.py |
最小方差无失真响应波束成形 |
- STFT 域:
(M, T, F, 2)— 通道, 时间帧, 频率, 实部/虚部 - 输出:
(N, T, F, 2)— 源信号, 时间帧, 频率, 实部/虚部 - 复数表示: 最后一维为 real/imag pair,用
torch.view_as_complex(X)转换
n_fft=1024,hop_length=512,sample_rate=16000- 窗函数:
sqrt(hann)
- 外部接口:1-indexed(
ref_mic=1表示第一个麦克风) - 内部计算:0-indexed(
ref_channel = ref_mic - 1) ref_mic=0表示禁用回投影
"mdp"— 最小失真原则"scale_constraint"— 参考通道尺度约束"none"— 不做投影回
- 遗忘因子 α(0.98~0.99),指数平均:
C = α*C_old + (1-α)*C_new - 每帧
n_iter通常为 1 reset()清除累积状态
- 在
src/bss/对应家族目录下创建文件(如src/bss/iva/new_algo.py) - 继承
BSSBase,用@register_bss("NEW_ALGO")装饰器注册:
from src.bss.base import BSSBase
from src.bss.registry import register_bss
@register_bss("NEW_ALGO")
class NewAlgo(BSSBase):
algorithm_family = "iva" # 或 "ilrma", "rcscme"
is_online = False # 或 True
def __init__(self, n_iter=100, ref_mic=1, **kwargs):
super().__init__(ref_mic=ref_mic)
self.n_iter = n_iter
def forward(self, X: torch.Tensor) -> torch.Tensor:
# X: (M, T, F, 2) → return: (N, T, F, 2)
...- 在家族
__init__.py中导入该类 - 在
src/bss/configs/algorithm/下创建对应 YAML 配置 - 添加测试到
test/
同理,继承 BFBase,用 @register_bf("NAME") 注册,配置放 src/bf/configs/algorithm/。
contrast_weights(r, contrast_func, gamma, eps)— 对比函数权重,支持:laplace, gaussian, logcosh, exp, pow1.5, pow0.5, powernmf_update(Tn, Vn, Y_n, eps)— ILRMA 共享的 NMF 乘法更新select_target_index(Y, sr, n_fft)— 通过语音频段 (300-3500Hz) 峰度选目标源
spatial_covariance_matrix(X, mask, eps)— 按频率估计空间协方差矩阵dominant_eigenvector(covariance)— 主特征向量steering_vector_from_covariance(cov, ref_channel, eps)— EVD 导向向量far_field_steering_vector(mic_positions, freqs, azimuth, ...)— 远场导向向量estimate_rank1_target_and_noise_covariances(cov, eps)— Rank-1 目标/噪声分解
STFT(win_len, shift_len, window)— STFT 变换/逆变换load_audio_sf(path, n_channels, seconds)→torch.Tensor (C, T)save_audio_sf(path, tensor, samplerate)
Hydra 分层配置:
configs/
├── config.yaml # defaults: algorithm, audio, output; seed, device
├── algorithm/*.yaml # 算法参数(name + params dict)
├── audio/default.yaml # input_file, sample_rate(16000), n_channels(3), n_fft(1024), hop_length(512)
└── output/default.yaml # output_dir("outputs"), save_audio(true)
BSS 和 BF 各自也有一套镜像的 src/bss/configs/ 和 src/bf/configs/。
# 全量测试
python -m pytest test/ -v
# 单个文件
python -m pytest test/test_mvdr.py -v
# 匹配关键字
python -m pytest test/ -k "registry" -v测试模式:
- 通过 registry 获取算法实例:
get_bss("NAME", **params) - 合成输入张量,检查输出 shape 和数值有限性
- 命名规范:文件
test_*.py,函数test_*()
核心:torch, torchaudio, numpy, scipy, soundfile, soxr, hydra-core, omegaconf
开发:pytest, pytest-cov
研究:matplotlib, pyroomacoustics
src/audio.py、src/utils.py是兼容层,实际实现在src/bss/和src/spatial.pyoutputs/目录已被 gitignore,但包含大量实验结果(pkf 文件、评估报告)tools/下的脚本是针对具体设备 & 场景的评估脚本,不属于核心库- ILRMA_SR 和 RCSCME 的
forward()签名与标准约定不同(额外参数) - 项目用
hatchling构建,wheel 只打包src/bss
IMPORTANT: This project has a knowledge graph. ALWAYS use the code-review-graph MCP tools BEFORE using Grep/Glob/Read to explore the codebase. The graph is faster, cheaper (fewer tokens), and gives you structural context (callers, dependents, test coverage) that file scanning cannot.
- Exploring code:
semantic_search_nodesorquery_graphinstead of Grep - Understanding impact:
get_impact_radiusinstead of manually tracing imports - Code review:
detect_changes+get_review_contextinstead of reading entire files - Finding relationships:
query_graphwith callers_of/callees_of/imports_of/tests_for - Architecture questions:
get_architecture_overview+list_communities
Fall back to Grep/Glob/Read only when the graph doesn't cover what you need.
| Tool | Use when |
|---|---|
detect_changes |
Reviewing code changes — gives risk-scored analysis |
get_review_context |
Need source snippets for review — token-efficient |
get_impact_radius |
Understanding blast radius of a change |
get_affected_flows |
Finding which execution paths are impacted |
query_graph |
Tracing callers, callees, imports, tests, dependencies |
semantic_search_nodes |
Finding functions/classes by name or keyword |
get_architecture_overview |
Understanding high-level codebase structure |
refactor_tool |
Planning renames, finding dead code |
- The graph auto-updates on file changes (via hooks).
- Use
detect_changesfor code review. - Use
get_affected_flowsto understand impact. - Use
query_graphpattern="tests_for" to check coverage.