EG/plugins/user/audio_reverb_effects/effects/modulation_processor.py
2025-12-12 16:16:15 +08:00

828 lines
27 KiB
Python

"""
调制效果处理器
实现专业的音频调制效果处理(合唱、镶边、移相等)
"""
import uuid
import math
import numpy as np
from typing import Dict, List, Any, Optional
class ModulationProcessor:
"""
调制效果处理器
实现专业的音频调制效果处理(合唱、镶边、移相等)
"""
def __init__(self, plugin):
"""
初始化调制效果处理器
Args:
plugin: 音频效果插件实例
"""
self.plugin = plugin
self.effects: Dict[str, Dict[str, Any]] = {}
self.stats = {
'total_effects': 0,
'active_effects': 0,
'effects_processed': 0,
'processing_time': 0.0,
'memory_usage': 0
}
self.buffer_size = plugin.buffer_size if plugin else 4096
self.sample_rate = 44100 # 默认采样率
def create_effect(self, parameters: Dict[str, Any]) -> str:
"""
创建调制效果
Args:
parameters: 调制效果参数
Returns:
效果ID
"""
try:
# 生成唯一ID
effect_id = str(uuid.uuid4())
# 默认参数
default_params = {
'type': 'chorus', # 效果类型: chorus, flanger, phaser
'rate': 1.0, # 速率 (Hz)
'depth': 0.5, # 深度 (0.0-1.0)
'feedback': 0.3, # 反馈 (-1.0-1.0)
'mix': 0.5, # 湿信号混合 (0.0-1.0)
'delay': 0.002, # 延迟时间 (秒)
'center_delay': 0.002, # 中心延迟 (秒)
'spread': 0.5, # 扩散 (0.0-1.0)
'stereo': True, # 立体声
'lfo_waveform': 'sine', # LFO波形: sine, triangle, square, sawtooth
'lfo_phase': 0.0, # LFO相位 (度)
'high_cut': 8000.0, # 高频截止 (Hz)
'low_cut': 100.0 # 低频截止 (Hz)
}
# 合并参数
effect_params = default_params.copy()
effect_params.update(parameters)
# 创建效果对象
effect = {
'id': effect_id,
'type': 'modulation',
'parameters': effect_params,
'active': True,
'buffers': {}, # 缓冲区
'lfo_phase': 0.0, # LFO相位状态
'delay_buffers': {}, # 延迟缓冲区
'filter_states': {}, # 滤波器状态
'created_time': self._get_current_time(),
'process_count': 0
}
# 初始化缓冲区
self._initialize_buffers(effect)
# 存储效果
self.effects[effect_id] = effect
# 更新统计信息
self.stats['total_effects'] += 1
self.stats['active_effects'] += 1
# 在插件中注册效果
if self.plugin:
self.plugin.effects[effect_id] = effect.copy()
print(f"✓ 调制效果创建成功: {effect_id}")
return effect_id
except Exception as e:
print(f"✗ 创建调制效果失败: {e}")
import traceback
traceback.print_exc()
return ""
def _initialize_buffers(self, effect: Dict[str, Any]):
"""
初始化缓冲区
Args:
effect: 效果对象
"""
effect_id = effect['id']
parameters = effect['parameters']
buffers = {}
# 输入和输出缓冲区
buffers['input'] = np.zeros(self.buffer_size)
buffers['output'] = np.zeros(self.buffer_size)
# 延迟缓冲区
max_delay_samples = int(0.1 * self.sample_rate) # 最大100ms延迟
buffers['delay_line'] = np.zeros(max_delay_samples)
buffers['delay_index'] = 0
# 立体声延迟缓冲区
if parameters.get('stereo', True):
buffers['delay_line_right'] = np.zeros(max_delay_samples)
buffers['delay_index_right'] = 0
# 滤波器状态
buffers['high_pass_state'] = np.zeros(2) # 双通道高通滤波器状态
buffers['low_pass_state'] = np.zeros(2) # 双通道低通滤波器状态
effect['buffers'] = buffers
def delete_effect(self, effect_id: str) -> bool:
"""
删除调制效果
Args:
effect_id: 效果ID
Returns:
是否删除成功
"""
if effect_id not in self.effects:
print(f"✗ 调制效果不存在: {effect_id}")
return False
try:
# 删除效果
del self.effects[effect_id]
# 更新统计信息
self.stats['active_effects'] -= 1
# 从插件中移除效果
if self.plugin and effect_id in self.plugin.effects:
del self.plugin.effects[effect_id]
print(f"✓ 调制效果已删除: {effect_id}")
return True
except Exception as e:
print(f"✗ 删除调制效果失败: {e}")
return False
def process(self, audio_data: Any, parameters: Dict[str, Any]) -> Any:
"""
处理调制效果
Args:
audio_data: 输入音频数据
parameters: 效果参数
Returns:
处理后的音频数据
"""
try:
import time
process_start = time.time()
# 如果输入是NumPy数组
if isinstance(audio_data, np.ndarray):
processed_data = self._process_numpy_array(audio_data, parameters)
else:
# 对于其他类型的数据,返回原始数据
processed_data = audio_data
# 更新统计信息
self.stats['effects_processed'] += 1
self.stats['processing_time'] += (time.time() - process_start)
return processed_data
except Exception as e:
print(f"✗ 调制效果处理失败: {e}")
return audio_data
def _process_numpy_array(self, audio_data: np.ndarray, parameters: Dict[str, Any]) -> np.ndarray:
"""
处理NumPy数组音频数据
Args:
audio_data: 输入音频数据
parameters: 效果参数
Returns:
处理后的音频数据
"""
# 获取参数
effect_type = parameters.get('type', 'chorus')
rate = max(0.01, min(10.0, parameters.get('rate', 1.0)))
depth = max(0.0, min(1.0, parameters.get('depth', 0.5)))
feedback = max(-1.0, min(1.0, parameters.get('feedback', 0.3)))
mix = max(0.0, min(1.0, parameters.get('mix', 0.5)))
delay = max(0.0001, min(0.05, parameters.get('delay', 0.002)))
center_delay = max(0.0001, min(0.05, parameters.get('center_delay', 0.002)))
spread = max(0.0, min(1.0, parameters.get('spread', 0.5)))
stereo = parameters.get('stereo', True)
lfo_waveform = parameters.get('lfo_waveform', 'sine')
high_cut = max(100.0, min(20000.0, parameters.get('high_cut', 8000.0)))
low_cut = max(20.0, min(10000.0, parameters.get('low_cut', 100.0)))
# 创建输出数组
if audio_data.ndim == 1:
# 单声道处理
output_data = np.zeros_like(audio_data)
# 初始化LFO相位
lfo_phase = 0.0
lfo_increment = 2.0 * math.pi * rate / self.sample_rate
for i in range(len(audio_data)):
sample = audio_data[i]
# 生成LFO值
lfo_value = self._generate_lfo_value(lfo_waveform, lfo_phase)
# 计算调制延迟
if effect_type == 'chorus':
mod_delay = center_delay + depth * 0.005 * lfo_value
elif effect_type == 'flanger':
mod_delay = delay + depth * 0.005 * lfo_value
else: # phaser
mod_delay = delay
# 限制延迟范围
mod_delay = max(0.0001, min(0.05, mod_delay))
# 读取延迟样本(简化插值)
delayed_sample = self._read_delayed_sample(sample, mod_delay, feedback)
# 应用滤波器(如果是镶边或移相器)
if effect_type in ['flanger', 'phaser']:
delayed_sample = self._apply_filter(delayed_sample, high_cut, low_cut)
# 混合干湿信号
output_data[i] = sample * (1.0 - mix) + delayed_sample * mix
# 更新LFO相位
lfo_phase += lfo_increment
if lfo_phase >= 2.0 * math.pi:
lfo_phase -= 2.0 * math.pi
else:
# 多声道处理(立体声)
output_data = np.zeros_like(audio_data)
# 初始化LFO相位
lfo_phase_left = 0.0
lfo_phase_right = parameters.get('lfo_phase', 0.0) * math.pi / 180.0 # 转换为弧度
lfo_increment = 2.0 * math.pi * rate / self.sample_rate
for channel in range(min(2, audio_data.shape[0])): # 最多处理两个声道
lfo_phase = lfo_phase_left if channel == 0 else lfo_phase_right
for i in range(audio_data.shape[1]):
sample = audio_data[channel, i]
# 生成LFO值
lfo_value = self._generate_lfo_value(lfo_waveform, lfo_phase)
# 计算调制延迟
if effect_type == 'chorus':
if channel == 0: # 左声道
mod_delay = center_delay + depth * 0.005 * lfo_value * (1.0 - spread)
else: # 右声道
mod_delay = center_delay + depth * 0.005 * lfo_value * (1.0 + spread)
elif effect_type == 'flanger':
mod_delay = delay + depth * 0.005 * lfo_value
else: # phaser
mod_delay = delay
# 限制延迟范围
mod_delay = max(0.0001, min(0.05, mod_delay))
# 读取延迟样本(简化插值)
delayed_sample = self._read_delayed_sample(sample, mod_delay, feedback)
# 应用滤波器(如果是镶边或移相器)
if effect_type in ['flanger', 'phaser']:
delayed_sample = self._apply_filter(delayed_sample, high_cut, low_cut)
# 混合干湿信号
output_data[channel, i] = sample * (1.0 - mix) + delayed_sample * mix
# 更新LFO相位
lfo_phase += lfo_increment
if lfo_phase >= 2.0 * math.pi:
lfo_phase -= 2.0 * math.pi
# 确保输出在有效范围内
processed_data = np.clip(output_data, -1.0, 1.0)
return processed_data
def _generate_lfo_value(self, waveform: str, phase: float) -> float:
"""
生成LFO值
Args:
waveform: 波形类型
phase: 相位(弧度)
Returns:
LFO值 (-1.0 到 1.0)
"""
if waveform == 'sine':
return math.sin(phase)
elif waveform == 'triangle':
# 三角波
normalized_phase = phase / (2.0 * math.pi)
if normalized_phase < 0.25:
return 4.0 * normalized_phase
elif normalized_phase < 0.75:
return 2.0 - 4.0 * normalized_phase
else:
return 4.0 * normalized_phase - 4.0
elif waveform == 'square':
# 方波
return 1.0 if math.sin(phase) >= 0.0 else -1.0
elif waveform == 'sawtooth':
# 锯齿波
normalized_phase = phase / (2.0 * math.pi)
return 2.0 * normalized_phase - 1.0
else:
# 默认使用正弦波
return math.sin(phase)
def _read_delayed_sample(self, input_sample: float, delay_time: float, feedback: float) -> float:
"""
读取延迟样本
Args:
input_sample: 输入样本
delay_time: 延迟时间(秒)
feedback: 反馈系数
Returns:
延迟样本
"""
# 这里使用简化的延迟处理
# 实际应用中会使用更复杂的延迟线和插值
# 计算延迟样本数
delay_samples = delay_time * self.sample_rate
# 简化的延迟效果(使用单一样本延迟)
delayed_sample = input_sample * 0.5 # 简化的延迟
# 应用反馈
if abs(feedback) > 0.01:
delayed_sample += delayed_sample * feedback * 0.3
return delayed_sample
def _apply_filter(self, sample: float, high_cut: float, low_cut: float) -> float:
"""
应用滤波器
Args:
sample: 输入样本
high_cut: 高频截止频率
low_cut: 低频截止频率
Returns:
滤波后的样本
"""
# 这里使用简化的滤波器
# 实际应用中会使用更复杂的滤波器设计
# 简单的一阶滤波器
filtered_sample = sample
# 高频截止(低通滤波器效果)
if high_cut < 10000.0:
filter_factor = high_cut / 10000.0
filtered_sample *= (0.5 + 0.5 * filter_factor)
# 低频截止(高通滤波器效果)
if low_cut > 100.0:
filter_factor = low_cut / 10000.0
filtered_sample *= (0.5 + 0.5 * filter_factor)
return filtered_sample
def set_parameter(self, effect_id: str, parameter: str, value: Any) -> bool:
"""
设置效果参数
Args:
effect_id: 效果ID
parameter: 参数名
value: 参数值
Returns:
是否设置成功
"""
if effect_id not in self.effects:
print(f"✗ 调制效果不存在: {effect_id}")
return False
try:
self.effects[effect_id]['parameters'][parameter] = value
# 如果是插件中的效果,也更新插件中的副本
if self.plugin and effect_id in self.plugin.effects:
self.plugin.effects[effect_id]['parameters'][parameter] = value
print(f"✓ 调制效果参数已设置: {effect_id}.{parameter} = {value}")
return True
except Exception as e:
print(f"✗ 设置调制效果参数失败: {e}")
return False
def get_parameters(self, effect_id: str) -> Dict[str, Any]:
"""
获取效果参数
Args:
effect_id: 效果ID
Returns:
参数字典
"""
if effect_id not in self.effects:
return {}
return self.effects[effect_id]['parameters'].copy()
def enable_effect(self, effect_id: str) -> bool:
"""
启用调制效果
Args:
effect_id: 效果ID
Returns:
是否启用成功
"""
if effect_id not in self.effects:
print(f"✗ 调制效果不存在: {effect_id}")
return False
try:
self.effects[effect_id]['active'] = True
self.stats['active_effects'] += 1
# 如果是插件中的效果,也更新插件中的副本
if self.plugin and effect_id in self.plugin.effects:
self.plugin.effects[effect_id]['active'] = True
print(f"✓ 调制效果已启用: {effect_id}")
return True
except Exception as e:
print(f"✗ 启用调制效果失败: {e}")
return False
def disable_effect(self, effect_id: str) -> bool:
"""
禁用调制效果
Args:
effect_id: 效果ID
Returns:
是否禁用成功
"""
if effect_id not in self.effects:
print(f"✗ 调制效果不存在: {effect_id}")
return False
try:
self.effects[effect_id]['active'] = False
self.stats['active_effects'] -= 1
# 如果是插件中的效果,也更新插件中的副本
if self.plugin and effect_id in self.plugin.effects:
self.plugin.effects[effect_id]['active'] = False
print(f"✓ 调制效果已禁用: {effect_id}")
return True
except Exception as e:
print(f"✗ 禁用调制效果失败: {e}")
return False
def cleanup(self):
"""清理所有资源"""
self.effects.clear()
self.stats = {
'total_effects': 0,
'active_effects': 0,
'effects_processed': 0,
'processing_time': 0.0,
'memory_usage': 0
}
print("✓ 调制效果处理器资源已清理")
def get_stats(self) -> Dict[str, int]:
"""
获取统计信息
Returns:
统计信息字典
"""
return self.stats.copy()
def _get_current_time(self) -> float:
"""
获取当前时间(秒)
Returns:
当前时间
"""
if self.plugin and self.plugin.plugin_manager and self.plugin.plugin_manager.world:
return self.plugin.plugin_manager.world.globalClock.getFrameTime()
import time
return time.time()
def update(self, dt: float):
"""
更新处理器状态
Args:
dt: 时间增量
"""
# 这里可以更新任何需要定期更新的状态
pass
def set_sample_rate(self, sample_rate: int):
"""
设置采样率
Args:
sample_rate: 采样率 (Hz)
"""
self.sample_rate = sample_rate
print(f"✓ 调制效果处理器采样率已设置: {sample_rate} Hz")
def set_rate(self, effect_id: str, rate: float) -> bool:
"""
设置LFO速率
Args:
effect_id: 效果ID
rate: 速率 (Hz)
Returns:
是否设置成功
"""
return self.set_parameter(effect_id, 'rate', max(0.01, min(10.0, rate)))
def set_depth(self, effect_id: str, depth: float) -> bool:
"""
设置深度
Args:
effect_id: 效果ID
depth: 深度 (0.0-1.0)
Returns:
是否设置成功
"""
return self.set_parameter(effect_id, 'depth', max(0.0, min(1.0, depth)))
def set_feedback(self, effect_id: str, feedback: float) -> bool:
"""
设置反馈
Args:
effect_id: 效果ID
feedback: 反馈 (-1.0-1.0)
Returns:
是否设置成功
"""
return self.set_parameter(effect_id, 'feedback', max(-1.0, min(1.0, feedback)))
def set_mix(self, effect_id: str, mix: float) -> bool:
"""
设置湿信号混合
Args:
effect_id: 效果ID
mix: 湿信号混合 (0.0-1.0)
Returns:
是否设置成功
"""
return self.set_parameter(effect_id, 'mix', max(0.0, min(1.0, mix)))
def set_delay(self, effect_id: str, delay: float) -> bool:
"""
设置延迟时间
Args:
effect_id: 效果ID
delay: 延迟时间 (秒)
Returns:
是否设置成功
"""
return self.set_parameter(effect_id, 'delay', max(0.0001, min(0.05, delay)))
def set_center_delay(self, effect_id: str, delay: float) -> bool:
"""
设置中心延迟时间
Args:
effect_id: 效果ID
delay: 中心延迟时间 (秒)
Returns:
是否设置成功
"""
return self.set_parameter(effect_id, 'center_delay', max(0.0001, min(0.05, delay)))
def set_spread(self, effect_id: str, spread: float) -> bool:
"""
设置扩散
Args:
effect_id: 效果ID
spread: 扩散 (0.0-1.0)
Returns:
是否设置成功
"""
return self.set_parameter(effect_id, 'spread', max(0.0, min(1.0, spread)))
def set_lfo_waveform(self, effect_id: str, waveform: str) -> bool:
"""
设置LFO波形
Args:
effect_id: 效果ID
waveform: 波形类型 (sine, triangle, square, sawtooth)
Returns:
是否设置成功
"""
valid_waveforms = ['sine', 'triangle', 'square', 'sawtooth']
if waveform not in valid_waveforms:
print(f"✗ 无效的LFO波形: {waveform}")
return False
return self.set_parameter(effect_id, 'lfo_waveform', waveform)
def enable_stereo(self, effect_id: str, enable: bool) -> bool:
"""
启用/禁用立体声
Args:
effect_id: 效果ID
enable: 是否启用
Returns:
是否设置成功
"""
return self.set_parameter(effect_id, 'stereo', enable)
def create_preset(self, preset_name: str) -> Dict[str, Any]:
"""
创建调制效果预设
Args:
preset_name: 预设名称
Returns:
预设参数字典
"""
presets = {
'chorus_light': {
'type': 'chorus',
'rate': 0.8,
'depth': 0.3,
'feedback': 0.2,
'mix': 0.4,
'center_delay': 0.004,
'spread': 0.3,
'lfo_waveform': 'sine'
},
'chorus_medium': {
'type': 'chorus',
'rate': 1.2,
'depth': 0.5,
'feedback': 0.3,
'mix': 0.6,
'center_delay': 0.006,
'spread': 0.5,
'lfo_waveform': 'sine'
},
'chorus_heavy': {
'type': 'chorus',
'rate': 1.8,
'depth': 0.7,
'feedback': 0.4,
'mix': 0.8,
'center_delay': 0.008,
'spread': 0.7,
'lfo_waveform': 'triangle'
},
'flanger_light': {
'type': 'flanger',
'rate': 0.5,
'depth': 0.4,
'feedback': 0.5,
'mix': 0.5,
'delay': 0.001,
'lfo_waveform': 'sine'
},
'flanger_medium': {
'type': 'flanger',
'rate': 0.8,
'depth': 0.6,
'feedback': 0.7,
'mix': 0.7,
'delay': 0.002,
'lfo_waveform': 'sine'
},
'flanger_heavy': {
'type': 'flanger',
'rate': 1.2,
'depth': 0.8,
'feedback': 0.9,
'mix': 0.9,
'delay': 0.003,
'lfo_waveform': 'triangle'
},
'phaser_light': {
'type': 'phaser',
'rate': 0.6,
'depth': 0.3,
'feedback': 0.2,
'mix': 0.4,
'delay': 0.0005,
'lfo_waveform': 'sine',
'high_cut': 6000.0,
'low_cut': 200.0
},
'phaser_medium': {
'type': 'phaser',
'rate': 1.0,
'depth': 0.5,
'feedback': 0.4,
'mix': 0.6,
'delay': 0.001,
'lfo_waveform': 'sine',
'high_cut': 8000.0,
'low_cut': 100.0
},
'phaser_heavy': {
'type': 'phaser',
'rate': 1.5,
'depth': 0.7,
'feedback': 0.6,
'mix': 0.8,
'delay': 0.002,
'lfo_waveform': 'triangle',
'high_cut': 10000.0,
'low_cut': 50.0
}
}
return presets.get(preset_name, {}).copy()
def get_lfo_meter(self, effect_id: str) -> float:
"""
获取LFO表
Args:
effect_id: 效果ID
Returns:
当前LFO值 (-1.0 到 1.0)
"""
if effect_id not in self.effects:
return 0.0
effect = self.effects[effect_id]
parameters = effect['parameters']
lfo_waveform = parameters.get('lfo_waveform', 'sine')
lfo_phase = effect.get('lfo_phase', 0.0)
return self._generate_lfo_value(lfo_waveform, lfo_phase)
def set_lfo_phase(self, effect_id: str, phase: float) -> bool:
"""
设置LFO相位
Args:
effect_id: 效果ID
phase: 相位 (度)
Returns:
是否设置成功
"""
# 转换为弧度并规范化
phase_rad = (phase * math.pi / 180.0) % (2.0 * math.pi)
return self.set_parameter(effect_id, 'lfo_phase', phase)