src/display/cli_display.py

import asyncio
import threading
import time
import os
from typing import Optional, Callable

from src.display.base_display import BaseDisplay
# 替换keyboard导入为pynput
from pynput import keyboard as pynput_keyboard

from src.utils.logging_config import get_logger

class CliDisplay(BaseDisplay):
    def __init__(self):
        super().__init__()  # 调用父类初始化
        """初始化CLI显示"""
        self.logger = get_logger(__name__)
        self.running = True

        # 状态相关
        self.current_status = "未连接"
        self.current_text = "待命"
        self.current_emotion = "😊"
        self.current_volume = 0  # 添加当前音量属性

        # 回调函数
        self.auto_callback = None
        self.status_callback = None
        self.text_callback = None
        self.emotion_callback = None
        self.abort_callback = None
        self.send_text_callback = None
        # 按键状态
        self.is_r_pressed = False

        # 状态缓存
        self.last_status = None
        self.last_text = None
        self.last_emotion = None
        self.last_volume = None

        # 键盘监听器
        self.keyboard_listener = None
        
        # 为异步操作添加事件循环
        self.loop = asyncio.new_event_loop()

    def set_callbacks(self,
                      press_callback: Optional[Callable] = None,
                      release_callback: Optional[Callable] = None,
                      status_callback: Optional[Callable] = None,
                      text_callback: Optional[Callable] = None,
                      emotion_callback: Optional[Callable] = None,
                      mode_callback: Optional[Callable] = None,
                      auto_callback: Optional[Callable] = None,
                      abort_callback: Optional[Callable] = None,
                      send_text_callback: Optional[Callable] = None):
        """设置回调函数"""
        self.status_callback = status_callback
        self.text_callback = text_callback
        self.emotion_callback = emotion_callback
        self.auto_callback = auto_callback
        self.abort_callback = abort_callback
        self.send_text_callback = send_text_callback

    def update_button_status(self, text: str):
        """更新按钮状态"""
        print(f"按钮状态: {text}")

    def update_status(self, status: str):
        """更新状态文本"""
        if status != self.current_status:
            self.current_status = status
            self._print_current_status()

    def update_text(self, text: str):
        """更新TTS文本"""
        if text != self.current_text:
            self.current_text = text
            self._print_current_status()

    def update_emotion(self, emotion_path: str):
        """更新表情
        emotion_path: GIF文件路径或表情字符串
        """
        if emotion_path != self.current_emotion:
            # 如果是gif文件路径，提取文件名作为表情名
            if emotion_path.endswith(".gif"):
                # 从路径中提取文件名，去掉.gif后缀
                emotion_name = os.path.basename(emotion_path).replace(".gif", "")
                self.current_emotion = f"[{emotion_name}]"
            else:
                # 如果不是gif路径，则直接使用
                self.current_emotion = emotion_path
            
            self._print_current_status()

    def start_keyboard_listener(self):
        """启动键盘监听"""
        try:
            def on_press(key):
                try:
                    # F2 按键处理 - 自动对话
                    if key == pynput_keyboard.Key.f2:
                        if self.auto_callback:
                            self.auto_callback()
                    # F3 按键处理 - 打断
                    elif key == pynput_keyboard.Key.f3:
                        if self.abort_callback:
                            self.abort_callback()
                except Exception as e:
                    self.logger.error(f"键盘事件处理错误: {e}")

            # 创建并启动监听器
            self.keyboard_listener = pynput_keyboard.Listener(
                on_press=on_press
            )
            self.keyboard_listener.start()
            self.logger.info("键盘监听器初始化成功")
        except Exception as e:
            self.logger.error(f"键盘监听器初始化失败: {e}")

    def stop_keyboard_listener(self):
        """停止键盘监听"""
        if self.keyboard_listener:
            try:
                self.keyboard_listener.stop()
                self.keyboard_listener = None
                self.logger.info("键盘监听器已停止")
            except Exception as e:
                self.logger.error(f"停止键盘监听器失败: {e}")

    def start(self):
        """启动CLI显示"""
        self._print_help()

        # 启动状态更新线程
        self.start_update_threads()

        # 启动键盘监听线程
        keyboard_thread = threading.Thread(target=self._keyboard_listener)
        keyboard_thread.daemon = True
        keyboard_thread.start()

        # 启动键盘监听
        self.start_keyboard_listener()

        # 主循环
        try:
            while self.running:
                time.sleep(0.1)
        except KeyboardInterrupt:
            self.on_close()

    def on_close(self):
        """关闭CLI显示"""
        self.running = False
        print("\n正在关闭应用...")
        self.stop_keyboard_listener()

    def _print_help(self):
        """打印帮助信息"""
        print("\n=== 小智Ai命令行控制 ===")
        print("可用命令：")
        print("  r     - 开始/停止对话")
        print("  x     - 打断当前对话")
        print("  s     - 显示当前状态")
        print("  v 数字 - 设置音量(0-100)")
        print("  q     - 退出程序")
        print("  h     - 显示此帮助信息")
        print("=====================\n")

    def _keyboard_listener(self):
        """键盘监听线程"""
        try:
            while self.running:
                cmd = input().lower().strip()
                if cmd == 'q':
                    self.on_close()
                    break
                elif cmd == 'h':
                    self._print_help()
                elif cmd == 'r':
                    if self.auto_callback:
                        self.auto_callback()
                elif cmd == 'x':
                    if self.abort_callback:
                        self.abort_callback()
                elif cmd == 's':
                    self._print_current_status()
                elif cmd.startswith('v '):  # 添加音量命令处理
                    try:
                        volume = int(cmd.split()[1])  # 获取音量值
                        if 0 <= volume <= 100:
                            self.update_volume(volume)
                            print(f"音量已设置为: {volume}%")
                        else:
                            print("音量必须在0-100之间")
                    except (IndexError, ValueError):
                        print("无效的音量值，格式：v <0-100>")
                else:
                    if self.send_text_callback:
                        # 获取应用程序的事件循环并在其中运行协程
                        from src.application import Application
                        app = Application.get_instance()
                        if app and app.loop:
                            asyncio.run_coroutine_threadsafe(
                                self.send_text_callback(cmd),
                                app.loop
                            )
                        else:
                            print("应用程序实例或事件循环不可用")
        except Exception as e:
            self.logger.error(f"键盘监听错误: {e}")

    def start_update_threads(self):
        """启动更新线程"""
        def update_loop():
            while self.running:
                try:
                    # 更新状态
                    if self.status_callback:
                        status = self.status_callback()
                        if status and status != self.current_status:
                            self.update_status(status)

                    # 更新文本
                    if self.text_callback:
                        text = self.text_callback()
                        if text and text != self.current_text:
                            self.update_text(text)

                    # 更新表情
                    if self.emotion_callback:
                        emotion = self.emotion_callback()
                        if emotion and emotion != self.current_emotion:
                            self.update_emotion(emotion)

                except Exception as e:
                    self.logger.error(f"状态更新错误: {e}")
                time.sleep(0.1)

        # 启动更新线程
        threading.Thread(target=update_loop, daemon=True).start()

    def _print_current_status(self):
        """打印当前状态"""
        # 检查是否有状态变化
        status_changed = (
            self.current_status != self.last_status or
            self.current_text != self.last_text or
            self.current_emotion != self.last_emotion or
            self.current_volume != self.last_volume
        )

        if status_changed:
            print("\n=== 当前状态 ===")
            print(f"状态: {self.current_status}")
            print(f"文本: {self.current_text}")
            print(f"表情: {self.current_emotion}")
            print(f"音量: {self.current_volume}%")
            print("===============\n")

            # 更新缓存
            self.last_status = self.current_status
            self.last_text = self.current_text
            self.last_emotion = self.current_emotion
            self.last_volume = self.current_volume