588 lines
22 KiB
Python
588 lines
22 KiB
Python
import os
|
|
import time
|
|
import logging
|
|
from PyQt5.QtCore import QObject, pyqtSignal
|
|
import platform
|
|
|
|
# 配置日志记录
|
|
logging.basicConfig(level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
|
logger = logging.getLogger("AudioFeatures")
|
|
|
|
class AudioFeatures(QObject):
|
|
"""音频功能模块，提供通话录音和音频播放功能"""
|
|
|
|
status_changed = pyqtSignal(str) # 状态变化信号
|
|
|
|
def __init__(self, lte_manager):
|
|
super().__init__()
|
|
self.lte_manager = lte_manager
|
|
self.recording = False
|
|
self.playing = False
|
|
self.auto_record_calls = False # 是否自动录制通话
|
|
self.auto_play_after_call = False # 是否在通话结束后自动播放录音
|
|
self.auto_play_on_answer = False # 是否在接听电话时自动播放声音
|
|
self.answer_play_audio_file = None # 接听时播放的音频文件
|
|
self.current_recording_file = None # 当前录音文件路径
|
|
|
|
# 获取用户 home 目录
|
|
self.user_home = os.path.expanduser("~")
|
|
|
|
# 创建默认存储路径：用户home/.LTE/REC/
|
|
self.base_storage_path = os.path.join(self.user_home, ".LTE")
|
|
self.storage_path = os.path.join(self.base_storage_path, "REC")
|
|
self.ensure_storage_path()
|
|
|
|
# 创建音频文件存储路径
|
|
self.audio_storage_path = os.path.join(self.base_storage_path, "AUDIO")
|
|
self.ensure_audio_storage_path()
|
|
|
|
# 支持的音频格式
|
|
self.supported_formats = [".amr", ".wav", ".mp3", ".pcm"]
|
|
|
|
def ensure_storage_path(self):
|
|
"""确保存储路径存在"""
|
|
try:
|
|
if not os.path.exists(self.storage_path):
|
|
os.makedirs(self.storage_path)
|
|
logger.info(f"已创建存储路径: {self.storage_path}")
|
|
except Exception as e:
|
|
logger.error(f"创建存储路径失败: {str(e)}")
|
|
|
|
def ensure_audio_storage_path(self):
|
|
"""确保音频文件存储路径存在"""
|
|
try:
|
|
if not os.path.exists(self.audio_storage_path):
|
|
os.makedirs(self.audio_storage_path)
|
|
logger.info(f"已创建音频存储路径: {self.audio_storage_path}")
|
|
except Exception as e:
|
|
logger.error(f"创建音频存储路径失败: {str(e)}")
|
|
|
|
def set_storage_path(self, path):
|
|
"""设置存储路径"""
|
|
if os.path.exists(path):
|
|
self.storage_path = path
|
|
logger.info(f"存储路径已设置为: {path}")
|
|
return True
|
|
else:
|
|
logger.error(f"存储路径不存在: {path}")
|
|
return False
|
|
|
|
def set_auto_record_calls(self, enabled):
|
|
"""设置是否自动录制通话"""
|
|
self.auto_record_calls = enabled
|
|
logger.info(f"自动录制通话功能已{'启用' if enabled else '禁用'}")
|
|
return True
|
|
|
|
def set_auto_play_after_call(self, enabled):
|
|
"""设置是否在通话结束后自动播放录音"""
|
|
self.auto_play_after_call = enabled
|
|
logger.info(f"通话结束后自动播放录音功能已{'启用' if enabled else '禁用'}")
|
|
return True
|
|
|
|
def set_auto_play_on_answer(self, enabled, audio_file=None):
|
|
"""
|
|
设置是否在接听电话时自动播放音频
|
|
|
|
参数:
|
|
- enabled: 是否启用该功能
|
|
- audio_file: 要播放的音频文件路径，如果为None则使用当前设置的文件
|
|
|
|
返回:
|
|
- bool: 是否成功设置
|
|
"""
|
|
self.auto_play_on_answer = enabled
|
|
|
|
if audio_file:
|
|
# 检查文件是否存在且格式支持
|
|
if os.path.exists(audio_file) and any(audio_file.lower().endswith(fmt) for fmt in self.supported_formats):
|
|
self.answer_play_audio_file = audio_file
|
|
logger.info(f"接听电话自动播放音频功能已{'启用' if enabled else '禁用'}, 音频文件: {os.path.basename(audio_file)}")
|
|
return True
|
|
else:
|
|
logger.error(f"音频文件不存在或格式不支持: {audio_file}")
|
|
return False
|
|
else:
|
|
# 如果未指定文件但已有设置的文件
|
|
if enabled and not self.answer_play_audio_file:
|
|
logger.warning("启用接听电话自动播放音频功能，但未指定音频文件")
|
|
return False
|
|
|
|
logger.info(f"接听电话自动播放音频功能已{'启用' if enabled else '禁用'}")
|
|
return True
|
|
|
|
def play_on_answer(self, phone_number=None):
|
|
"""
|
|
接听电话时自动播放音频
|
|
|
|
参数:
|
|
- phone_number: 可选，来电号码用于日志记录
|
|
|
|
返回:
|
|
- bool: 是否成功播放
|
|
"""
|
|
if not self.auto_play_on_answer or not self.answer_play_audio_file:
|
|
logger.info("接听电话自动播放功能未启用或未设置音频文件")
|
|
return False
|
|
|
|
if not os.path.exists(self.answer_play_audio_file):
|
|
logger.error(f"接听自动播放音频文件不存在: {self.answer_play_audio_file}")
|
|
return False
|
|
|
|
# 先停止可能正在播放的音频
|
|
if self.playing:
|
|
self.stop_audio()
|
|
time.sleep(0.5) # 等待停止完成
|
|
|
|
# 使用远程播放模式，让对方能听到声音
|
|
logger.info(f"接听电话({phone_number})，自动播放音频: {os.path.basename(self.answer_play_audio_file)}")
|
|
return self.play_audio(self.answer_play_audio_file, play_path=1) # play_path=1表示远程播放，对方听得到
|
|
|
|
def start_call_recording(self, phone_number=None):
|
|
"""
|
|
开始通话录音
|
|
|
|
参数:
|
|
- phone_number: 电话号码，用于文件命名
|
|
|
|
返回:
|
|
- bool: 是否成功开始录音
|
|
- str: 录音文件路径
|
|
"""
|
|
if not phone_number:
|
|
phone_number = self.lte_manager.call_number if hasattr(self.lte_manager, 'call_number') else "unknown"
|
|
|
|
# 创建基于电话号码和时间的文件名
|
|
timestamp = time.strftime("%Y%m%d_%H%M%S")
|
|
filename = f"call_{phone_number}_{timestamp}.wav"
|
|
|
|
# 完整文件路径
|
|
file_path = os.path.join(self.storage_path, filename)
|
|
|
|
# 存储当前录音文件路径用于可能的自动播放
|
|
self.current_recording_file = file_path
|
|
|
|
# 调用录音方法，使用双方声音混合录制模式(3)
|
|
result = self.start_recording(filename=filename, record_path=3)
|
|
|
|
return result, file_path
|
|
|
|
def start_recording(self, filename=None, record_path=1):
|
|
"""
|
|
开始录音
|
|
|
|
参数:
|
|
- filename: 录音文件名，不包含路径。如果未提供，则使用时间戳命名
|
|
- record_path: 录音路径类型
|
|
1 = 本地路径 (录制本地麦克风)
|
|
2 = 远程路径 (录制通话对方声音)
|
|
3 = 混合模式 (录制双方声音)
|
|
|
|
返回:
|
|
- bool: 是否成功开始录音
|
|
"""
|
|
if self.recording:
|
|
logger.warning("当前已有录音正在进行")
|
|
return False
|
|
|
|
if not self.lte_manager.is_connected():
|
|
logger.error("未连接到LTE模块，无法开始录音")
|
|
self.status_changed.emit("未连接到LTE模块，无法开始录音")
|
|
return False
|
|
|
|
# 如果未提供文件名，使用时间戳命名
|
|
if not filename:
|
|
timestamp = time.strftime("%Y%m%d_%H%M%S")
|
|
filename = f"recording_{timestamp}.wav" # 默认使用wav格式
|
|
|
|
# 确保文件格式正确
|
|
if not any(filename.lower().endswith(fmt) for fmt in self.supported_formats):
|
|
filename += ".wav" # 默认使用wav格式
|
|
|
|
# 组合完整路径
|
|
file_path = os.path.join(self.storage_path, filename)
|
|
# 使用模块内路径格式 (c:/ 对应模块内存储)
|
|
module_path = f"c:/{os.path.basename(filename)}"
|
|
|
|
# 记录当前录音文件路径
|
|
self.current_recording_file = file_path
|
|
|
|
# 发送录音命令
|
|
command = f'AT+CREC={record_path},"{module_path}"'
|
|
response = self.lte_manager.send_at_command(command)
|
|
|
|
if "+CREC: 1" in response or "+CREC: 2" in response or "+CREC: 3" in response:
|
|
self.recording = True
|
|
logger.info(f"录音已开始: {file_path}")
|
|
self.status_changed.emit(f"录音已开始: {os.path.basename(file_path)}")
|
|
return True
|
|
else:
|
|
logger.error(f"开始录音失败: {response}")
|
|
self.status_changed.emit(f"开始录音失败")
|
|
self.current_recording_file = None
|
|
return False
|
|
|
|
def stop_recording(self):
|
|
"""
|
|
停止录音
|
|
|
|
返回:
|
|
- bool: 是否成功停止录音
|
|
"""
|
|
if not self.recording:
|
|
logger.warning("当前没有录音正在进行")
|
|
return False
|
|
|
|
if not self.lte_manager.is_connected():
|
|
logger.error("未连接到LTE模块，无法停止录音")
|
|
self.status_changed.emit("未连接到LTE模块，无法停止录音")
|
|
return False
|
|
|
|
# 发送停止录音命令
|
|
command = "AT+CREC=0"
|
|
response = self.lte_manager.send_at_command(command)
|
|
|
|
if "+CREC: 0" in response:
|
|
self.recording = False
|
|
logger.info("录音已停止")
|
|
self.status_changed.emit("录音已停止")
|
|
|
|
# 等待模块完成录音处理
|
|
time.sleep(0.5)
|
|
|
|
# 检查是否收到录音完成信号
|
|
if "+CREC: crec stop" in response:
|
|
logger.info("录音已完成处理")
|
|
|
|
# 检查是否需要自动播放录音
|
|
if self.auto_play_after_call and self.current_recording_file:
|
|
logger.info(f"准备自动播放录音: {self.current_recording_file}")
|
|
# 等待一下以确保录音文件已完成保存
|
|
time.sleep(1)
|
|
self.play_audio(self.current_recording_file)
|
|
|
|
return True
|
|
else:
|
|
logger.error(f"停止录音失败: {response}")
|
|
self.status_changed.emit("停止录音失败")
|
|
return False
|
|
|
|
def is_recording(self):
|
|
"""
|
|
检查是否正在录音
|
|
|
|
返回:
|
|
- bool: 是否正在录音
|
|
"""
|
|
if not self.lte_manager.is_connected():
|
|
return False
|
|
|
|
# 查询录音状态
|
|
response = self.lte_manager.send_at_command("AT+CREC?")
|
|
|
|
if "+CREC: 1" in response or "+CREC: 2" in response or "+CREC: 3" in response:
|
|
self.recording = True
|
|
return True
|
|
else:
|
|
self.recording = False
|
|
return False
|
|
|
|
def play_audio(self, filename, repeat=0, play_path=0):
|
|
"""
|
|
播放音频文件
|
|
|
|
参数:
|
|
- filename: 音频文件名，可以是绝对路径或相对路径
|
|
- repeat: 重复播放次数，0表示只播放一次，1-255表示重复播放的次数
|
|
- play_path: 播放路径
|
|
0 = 本地播放(默认)
|
|
1 = 远程播放(通话时对方听到)
|
|
2 = 双方都播放(本地和远程)
|
|
|
|
返回:
|
|
- bool: 是否成功开始播放
|
|
"""
|
|
if self.playing:
|
|
logger.warning("当前已有音频正在播放")
|
|
self.stop_audio() # 先停止当前播放
|
|
|
|
if not self.lte_manager.is_connected():
|
|
logger.error("未连接到LTE模块，无法播放音频")
|
|
self.status_changed.emit("未连接到LTE模块，无法播放音频")
|
|
return False
|
|
|
|
# 确保文件存在且格式受支持
|
|
if not os.path.exists(filename):
|
|
# 尝试在存储路径中查找
|
|
full_path = os.path.join(self.storage_path, filename)
|
|
if not os.path.exists(full_path):
|
|
# 尝试在音频路径中查找
|
|
audio_path = os.path.join(self.audio_storage_path, filename)
|
|
if not os.path.exists(audio_path):
|
|
logger.error(f"音频文件不存在: {filename}")
|
|
self.status_changed.emit(f"音频文件不存在: {os.path.basename(filename)}")
|
|
return False
|
|
filename = audio_path
|
|
else:
|
|
filename = full_path
|
|
|
|
if not any(filename.lower().endswith(fmt) for fmt in self.supported_formats):
|
|
logger.error(f"不支持的音频格式: {filename}")
|
|
self.status_changed.emit(f"不支持的音频格式: {os.path.basename(filename)}")
|
|
return False
|
|
|
|
# 转换为模块内路径格式
|
|
module_path = f"c:/{os.path.basename(filename)}"
|
|
|
|
# 根据文件类型选择播放命令 (对于wav文件可以使用AT+CCMXPLAYWAV)
|
|
if filename.lower().endswith(".wav"):
|
|
command = f'AT+CCMXPLAYWAV="{module_path}",{play_path}'
|
|
else:
|
|
# 其他格式使用AT+CCMXPLAY命令
|
|
command = f'AT+CCMXPLAY="{module_path}",{play_path},{repeat}'
|
|
|
|
response = self.lte_manager.send_at_command(command)
|
|
|
|
success = False
|
|
if "+CCMXPLAY:" in response and "OK" in response:
|
|
success = True
|
|
elif "+CCMXPLAYWAV:" in response and "OK" in response:
|
|
success = True
|
|
|
|
if success:
|
|
self.playing = True
|
|
logger.info(f"开始播放音频: {filename}")
|
|
self.status_changed.emit(f"开始播放音频: {os.path.basename(filename)}")
|
|
return True
|
|
else:
|
|
logger.error(f"播放音频失败: {response}")
|
|
self.status_changed.emit("播放音频失败")
|
|
return False
|
|
|
|
def stop_audio(self):
|
|
"""
|
|
停止音频播放
|
|
|
|
返回:
|
|
- bool: 是否成功停止播放
|
|
"""
|
|
if not self.playing:
|
|
logger.info("当前没有音频播放")
|
|
return True
|
|
|
|
if not self.lte_manager.is_connected():
|
|
logger.error("未连接到LTE模块，无法停止播放")
|
|
self.status_changed.emit("未连接到LTE模块，无法停止播放")
|
|
return False
|
|
|
|
# 检查是否正在播放WAV文件
|
|
response_wav = self.lte_manager.send_at_command("AT+CCMXSTOPWAV")
|
|
response_normal = self.lte_manager.send_at_command("AT+CCMXSTOP")
|
|
|
|
success = False
|
|
|
|
if "+CCMXSTOPWAV:" in response_wav and "OK" in response_wav:
|
|
success = True
|
|
elif "+CCMXSTOP:" in response_normal and "OK" in response_normal:
|
|
success = True
|
|
|
|
if success:
|
|
self.playing = False
|
|
logger.info("音频播放已停止")
|
|
self.status_changed.emit("音频播放已停止")
|
|
return True
|
|
else:
|
|
logger.error(f"停止音频播放失败: {response_wav}, {response_normal}")
|
|
self.status_changed.emit("停止音频播放失败")
|
|
return False
|
|
|
|
def set_ringtone(self, filename):
|
|
"""
|
|
设置来电铃声
|
|
|
|
参数:
|
|
- filename: 铃声文件名，可以是绝对路径或相对路径
|
|
|
|
返回:
|
|
- bool: 是否成功设置铃声
|
|
"""
|
|
if not self.lte_manager.is_connected():
|
|
logger.error("未连接到LTE模块，无法设置铃声")
|
|
self.status_changed.emit("未连接到LTE模块，无法设置铃声")
|
|
return False
|
|
|
|
# 确保文件存在且格式受支持
|
|
if not os.path.exists(filename):
|
|
# 尝试在存储路径中查找
|
|
full_path = os.path.join(self.storage_path, filename)
|
|
if not os.path.exists(full_path):
|
|
# 尝试在音频路径中查找
|
|
audio_path = os.path.join(self.audio_storage_path, filename)
|
|
if not os.path.exists(audio_path):
|
|
logger.error(f"铃声文件不存在: {filename}")
|
|
self.status_changed.emit(f"铃声文件不存在: {os.path.basename(filename)}")
|
|
return False
|
|
filename = audio_path
|
|
else:
|
|
filename = full_path
|
|
|
|
if not any(filename.lower().endswith(fmt) for fmt in self.supported_formats):
|
|
logger.error(f"不支持的铃声格式: {filename}")
|
|
self.status_changed.emit(f"不支持的铃声格式: {os.path.basename(filename)}")
|
|
return False
|
|
|
|
# 转换为模块内路径格式
|
|
module_path = f"c:/{os.path.basename(filename)}"
|
|
|
|
# 发送设置铃声命令
|
|
command = f'AT+CRINGSET="{module_path}",1'
|
|
response = self.lte_manager.send_at_command(command)
|
|
|
|
if "OK" in response:
|
|
logger.info(f"铃声已设置: {filename}")
|
|
self.status_changed.emit(f"铃声已设置: {os.path.basename(filename)}")
|
|
return True
|
|
else:
|
|
logger.error(f"设置铃声失败: {response}")
|
|
self.status_changed.emit("设置铃声失败")
|
|
return False
|
|
|
|
def ring_switch(self, enable=True):
|
|
"""
|
|
开启或关闭铃声
|
|
|
|
参数:
|
|
- enable: 是否启用铃声
|
|
|
|
返回:
|
|
- bool: 是否成功切换铃声状态
|
|
"""
|
|
if not self.lte_manager.is_connected():
|
|
logger.error("未连接到LTE模块，无法切换铃声状态")
|
|
self.status_changed.emit("未连接到LTE模块，无法切换铃声状态")
|
|
return False
|
|
|
|
# 发送铃声开关命令
|
|
status = 1 if enable else 0
|
|
command = f"AT+CRTSWITCH={status}"
|
|
response = self.lte_manager.send_at_command(command)
|
|
|
|
if "OK" in response:
|
|
state = "开启" if enable else "关闭"
|
|
logger.info(f"铃声已{state}")
|
|
self.status_changed.emit(f"铃声已{state}")
|
|
return True
|
|
else:
|
|
logger.error(f"切换铃声状态失败: {response}")
|
|
self.status_changed.emit("切换铃声状态失败")
|
|
return False
|
|
|
|
def generate_dtmf(self, dtmf_string, duration=1, time_base=100):
|
|
"""
|
|
生成DTMF音
|
|
|
|
参数:
|
|
- dtmf_string: DTMF字符串，如"1,2,3,4"
|
|
- duration: 持续时间因子，1-100
|
|
- time_base: 时间基准，50-500ms
|
|
|
|
返回:
|
|
- bool: 是否成功生成DTMF音
|
|
"""
|
|
if not self.lte_manager.is_connected():
|
|
logger.error("未连接到LTE模块，无法生成DTMF音")
|
|
self.status_changed.emit("未连接到LTE模块，无法生成DTMF音")
|
|
return False
|
|
|
|
# 确保DTMF字符串格式正确（数字、字母A-D、*、#，逗号分隔）
|
|
valid_chars = set("0123456789ABCD*#,")
|
|
if not all(c.upper() in valid_chars for c in dtmf_string):
|
|
logger.error(f"无效的DTMF字符串: {dtmf_string}")
|
|
self.status_changed.emit("无效的DTMF字符串")
|
|
return False
|
|
|
|
# 发送DTMF生成命令
|
|
command = f'AT+CLDTMF={duration},"{dtmf_string}",{time_base},0'
|
|
response = self.lte_manager.send_at_command(command)
|
|
|
|
if "OK" in response:
|
|
logger.info(f"DTMF音已生成: {dtmf_string}")
|
|
self.status_changed.emit(f"DTMF音已生成")
|
|
return True
|
|
else:
|
|
logger.error(f"生成DTMF音失败: {response}")
|
|
self.status_changed.emit("生成DTMF音失败")
|
|
return False
|
|
|
|
def generate_tone(self, frequency=1000, period_on=200, period_off=200, duration=1000):
|
|
"""
|
|
生成特定频率的音调
|
|
|
|
参数:
|
|
- frequency: 频率，20-4000Hz
|
|
- period_on: 音调开启周期，50-25500ms
|
|
- period_off: 音调关闭周期，0或40-25500ms
|
|
- duration: 持续时间，50-500000ms
|
|
|
|
返回:
|
|
- bool: 是否成功生成音调
|
|
"""
|
|
if not self.lte_manager.is_connected():
|
|
logger.error("未连接到LTE模块，无法生成音调")
|
|
self.status_changed.emit("未连接到LTE模块，无法生成音调")
|
|
return False
|
|
|
|
# 检查参数范围
|
|
if not (20 <= frequency <= 4000):
|
|
logger.error(f"频率超出范围(20-4000Hz): {frequency}")
|
|
frequency = max(20, min(frequency, 4000))
|
|
|
|
if not (50 <= period_on <= 25500):
|
|
logger.error(f"开启周期超出范围(50-25500ms): {period_on}")
|
|
period_on = max(50, min(period_on, 25500))
|
|
|
|
if period_off != 0 and not (40 <= period_off <= 25500):
|
|
logger.error(f"关闭周期超出范围(0或40-25500ms): {period_off}")
|
|
period_off = max(40, min(period_off, 25500))
|
|
|
|
if not (50 <= duration <= 500000):
|
|
logger.error(f"持续时间超出范围(50-500000ms): {duration}")
|
|
duration = max(50, min(duration, 500000))
|
|
|
|
# 发送音调生成命令
|
|
command = f"AT+SIMTONE=1,{frequency},{period_on},{period_off},{duration}"
|
|
response = self.lte_manager.send_at_command(command)
|
|
|
|
if "OK" in response:
|
|
logger.info(f"音调已生成: {frequency}Hz")
|
|
self.status_changed.emit(f"音调已生成: {frequency}Hz")
|
|
return True
|
|
else:
|
|
logger.error(f"生成音调失败: {response}")
|
|
self.status_changed.emit("生成音调失败")
|
|
return False
|
|
|
|
def stop_tone(self):
|
|
"""
|
|
停止音调
|
|
|
|
返回:
|
|
- bool: 是否成功停止音调
|
|
"""
|
|
if not self.lte_manager.is_connected():
|
|
logger.error("未连接到LTE模块，无法停止音调")
|
|
self.status_changed.emit("未连接到LTE模块，无法停止音调")
|
|
return False
|
|
|
|
# 发送停止音调命令
|
|
command = "AT+SIMTONE=0"
|
|
response = self.lte_manager.send_at_command(command)
|
|
|
|
if "OK" in response:
|
|
logger.info("音调已停止")
|
|
self.status_changed.emit("音调已停止")
|
|
return True
|
|
else:
|
|
logger.error(f"停止音调失败: {response}")
|
|
self.status_changed.emit("停止音调失败")
|
|
return False |