chore: import R0nY3n/LTE_manager main snapshot

This commit is contained in:
2026-07-20 16:35:15 +00:00
commit b8438bba61
48 changed files with 15676 additions and 0 deletions
+588
View File
@@ -0,0 +1,588 @@
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