62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
import asyncio
|
|
from dataclasses import dataclass
|
|
from typing import Dict, Optional
|
|
|
|
|
|
VOICE_TARGET_KIND_DEVICE = "device"
|
|
VOICE_TARGET_KIND_PARENT = "parent"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class VoiceTarget:
|
|
kind: str
|
|
target_device_id: Optional[str] = None
|
|
uuid: Optional[str] = None
|
|
audio_cache_key: Optional[str] = None
|
|
|
|
|
|
class DeviceTargetCache:
|
|
def __init__(self):
|
|
self.targets: Dict[str, VoiceTarget] = {}
|
|
self.lock = asyncio.Lock()
|
|
|
|
async def get_target(self, device_id: str) -> Optional[str]:
|
|
async with self.lock:
|
|
target = self.targets.get(device_id)
|
|
if not target or target.kind != VOICE_TARGET_KIND_DEVICE:
|
|
return None
|
|
return target.target_device_id
|
|
|
|
async def get_voice_target(self, device_id: str) -> Optional[VoiceTarget]:
|
|
async with self.lock:
|
|
return self.targets.get(device_id)
|
|
|
|
async def set_target(self, device_id: str, target_device_id: str):
|
|
async with self.lock:
|
|
self.targets[device_id] = VoiceTarget(
|
|
kind=VOICE_TARGET_KIND_DEVICE,
|
|
target_device_id=target_device_id,
|
|
audio_cache_key=target_device_id,
|
|
)
|
|
|
|
async def set_parent_target(self, device_id: str, uuid: str):
|
|
async with self.lock:
|
|
self.targets[device_id] = VoiceTarget(
|
|
kind=VOICE_TARGET_KIND_PARENT,
|
|
uuid=uuid,
|
|
audio_cache_key=f"parent:{device_id}",
|
|
)
|
|
|
|
async def remove_target(self, device_id: str):
|
|
async with self.lock:
|
|
if device_id in self.targets:
|
|
del self.targets[device_id]
|
|
|
|
async def get_all_targets(self) -> list:
|
|
async with self.lock:
|
|
return list(self.targets.items())
|
|
|
|
|
|
# 单例实例
|
|
device_target_cache = DeviceTargetCache()
|