cmvr-es-cli/cmvr/speaker_client.py

import grpc
from typing import Tuple

from .enums import CMVRErrorCode
from .models import SpeakerState

class SpeakerClient:
    """扬声器客户端"""

    def __init__(self, device_id: str, stub):
        self.device_id = device_id
        self.stub = stub
        self.generated = None  # 推迟导入

    def _import_generated(self):
        """推迟导入 generated"""
        if self.generated is None:
            try:
                # 从正确的路径导入生成的模块
                from generated.cmvr.api import common_pb2, speaker_command_pb2
                self.generated = type('GeneratedModules', (), {
                    'common_pb2': common_pb2,
                    'speaker_command_pb2': speaker_command_pb2
                })
            except ImportError as e:
                print(f"导入生成的模块失败: {e}")
                print("请确保已生成 protobuf 代码")
                raise
        return self.generated

    def _create_command_header(self):
        """创建命令头"""
        generated = self._import_generated()
        header = generated.common_pb2.CommandHeader.Request()
        header.device_id = self.device_id
        header.timestamp.GetCurrentTime()
        return header

    def get_status(self) -> Tuple[CMVRErrorCode, SpeakerState]:
        """获取扬声器状态"""
        try:
            generated = self._import_generated()
            request = generated.speaker_command_pb2.GetSpeakerStateCommand.Request()
            request.header.CopyFrom(self._create_command_header())

            response = self.stub.GetStatus(request)

            if not response.header.success:
                return CMVRErrorCode.CMVR_RPC_FAILED, SpeakerState()

            state = SpeakerState(
                is_initialized=response.state.is_initialized,
                is_running=response.state.is_running,
                is_decoding=response.state.is_decoding,
                is_paused=response.state.is_paused,
                volume=response.state.volume
            )

            return CMVRErrorCode.CMVR_SUCCESS, state

        except grpc.RpcError as e:
            print(f"获取扬声器状态失败: {e}")
            return CMVRErrorCode.CMVR_RPC_FAILED, SpeakerState()

    def play_audio(self, audio_path: str) -> CMVRErrorCode:
        """播放音频"""
        try:
            generated = self._import_generated()
            request = generated.speaker_command_pb2.PlayAudioCommand.Request()
            request.header.CopyFrom(self._create_command_header())
            request.audio_path = audio_path

            response = self.stub.PlayAudio(request)

            return CMVRErrorCode.CMVR_SUCCESS if response.header.success else CMVRErrorCode.CMVR_RPC_FAILED

        except grpc.RpcError as e:
            print(f"播放音频失败: {e}")
            return CMVRErrorCode.CMVR_RPC_FAILED

    def pause_audio(self) -> CMVRErrorCode:
        """暂停播放"""
        try:
            generated = self._import_generated()
            request = generated.speaker_command_pb2.PauseSpeakerCommand.Request()
            request.header.CopyFrom(self._create_command_header())

            response = self.stub.PausePlayback(request)

            return CMVRErrorCode.CMVR_SUCCESS if response.header.success else CMVRErrorCode.CMVR_RPC_FAILED

        except grpc.RpcError as e:
            print(f"暂停播放失败: {e}")
            return CMVRErrorCode.CMVR_RPC_FAILED

    def resume_audio(self) -> CMVRErrorCode:
        """继续播放"""
        try:
            generated = self._import_generated()
            request = generated.speaker_command_pb2.ResumeSpeakerCommand.Request()
            request.header.CopyFrom(self._create_command_header())

            response = self.stub.ResumePlayback(request)

            return CMVRErrorCode.CMVR_SUCCESS if response.header.success else CMVRErrorCode.CMVR_RPC_FAILED

        except grpc.RpcError as e:
            print(f"继续播放失败: {e}")
            return CMVRErrorCode.CMVR_RPC_FAILED

    def stop_audio(self) -> CMVRErrorCode:
        """停止播放"""
        try:
            generated = self._import_generated()
            request = generated.speaker_command_pb2.StopSpeakerCommand.Request()
            request.header.CopyFrom(self._create_command_header())

            response = self.stub.StopPlayback(request)

            return CMVRErrorCode.CMVR_SUCCESS if response.header.success else CMVRErrorCode.CMVR_RPC_FAILED

        except grpc.RpcError as e:
            print(f"停止播放失败: {e}")
            return CMVRErrorCode.CMVR_RPC_FAILED

    def set_volume(self, volume: float) -> CMVRErrorCode:
        """设置音量"""
        try:
            generated = self._import_generated()
            request = generated.speaker_command_pb2.SetSpeakerVolumeCommand.Request()
            request.header.CopyFrom(self._create_command_header())
            request.volume = volume

            response = self.stub.SetVolume(request)

            return CMVRErrorCode.CMVR_SUCCESS if response.header.success else CMVRErrorCode.CMVR_RPC_FAILED

        except grpc.RpcError as e:
            print(f"设置音量失败: {e}")
            return CMVRErrorCode.CMVR_RPC_FAILED

    def get_volume(self) -> Tuple[CMVRErrorCode, float]:
        """获取音量"""
        try:
            generated = self._import_generated()
            request = generated.speaker_command_pb2.GetSpeakerVolumeCommand.Request()
            request.header.CopyFrom(self._create_command_header())

            response = self.stub.GetVolume(request)

            if not response.header.success:
                return CMVRErrorCode.CMVR_RPC_FAILED, 0.0

            return CMVRErrorCode.CMVR_SUCCESS, response.volume

        except grpc.RpcError as e:
            print(f"获取音量失败: {e}")
            return CMVRErrorCode.CMVR_RPC_FAILED, 0.0