| import uuid |
|
|
| from voice_dialogue.utils.cache import LRUCacheDict |
| from .enums import AudioState |
|
|
|
|
| class VoiceStateManager: |
| """语音状态管理器""" |
|
|
| def __init__(self): |
| self._task_id = '' |
| self._audio_task_states = LRUCacheDict(maxsize=10) |
| self.waiting_second_answer_mapping = LRUCacheDict(maxsize=10) |
| self._interrupt_task_id = '' |
|
|
| @property |
| def task_id(self): |
| return self._task_id |
|
|
| @task_id.setter |
| def task_id(self, value): |
| self._task_id = value |
|
|
| def create_task_id(self): |
| """创建新的任务ID""" |
| self._task_id = f'{uuid.uuid4()}' |
|
|
| def reset_task_id(self): |
| """重置任务ID""" |
| self._task_id = '' |
|
|
| def get_audio_task_state(self, task_id): |
| """获取音频任务状态""" |
| return self._audio_task_states.get(task_id) |
|
|
| def set_audio_playing(self, task_id): |
| """设置音频为播放状态""" |
| self._audio_task_states[task_id] = AudioState.PLAYING |
|
|
| def drop_audio_task(self, task_id): |
| """丢弃音频任务""" |
| self._audio_task_states[task_id] = AudioState.DROP |
|
|
| def cleanup_task_state(self, task_id): |
| """清理任务状态""" |
| if task_id in self._audio_task_states: |
| del self._audio_task_states[task_id] |
|
|
| @property |
| def interrupt_task_id(self): |
| return self._interrupt_task_id |
|
|
| @interrupt_task_id.setter |
| def interrupt_task_id(self, value): |
| self._interrupt_task_id = value |
|
|
