File size: 3,494 Bytes
ad88834 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 | import asyncio
import os
import time
from collections import deque
from watcherobot.application import ApplicationContext
from watcherobot.application.context import ApplicationEnvironmentError
from watcherobot.application.transport import DaemonApplicationTransport
from watcherobot.errors import WatcheRobotError
class DeviceTransport(DaemonApplicationTransport):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.device_ready = asyncio.Event()
self.audio_sender_lock = asyncio.Lock()
self.events = deque(maxlen=128)
self.add_message_listener(self.observe_ready)
def observe_ready(self, message):
if message["type"] in {
"evt.audio.buffer_status", "evt.microphone.state", "evt.sdk.ready",
"evt.sdk.resource_snapshot", "sys.nack",
}:
self.events.append({"time": time.monotonic(), **message})
if message["type"] == "evt.sdk.ready":
self.device_ready.set()
def start(self, *, timeout=15):
super().start(timeout=timeout)
@property
def closed(self):
"""应用与守护进程之间的通道已经结束，之后的设备指令都无法执行。"""
thread = self._thread
return thread is not None and not thread.is_alive()
async def _on_channels_connected(self):
self.device_ready.clear()
async with asyncio.timeout(self.command_timeout):
await self._send_command("sys.sdk.ready.get", {}, self.command_timeout)
await self.device_ready.wait()
self._started_event.set()
async def _send_audio_stream(self, pcm, stream_id, chunk_bytes):
# 上一段发送任务完成取消清理后，下一段才可以修改共享流控状态。
async with self.audio_sender_lock:
await super()._send_audio_stream(pcm, stream_id, chunk_bytes)
async def _take_audio_credit(self, stream_id):
condition = self._audio_credit_condition
if condition is None:
raise WatcheRobotError("audio flow control is not initialized")
# 在发送任务内部等待，取消时同时结束 Condition 等待。
async with asyncio.timeout(self.command_timeout), condition:
await condition.wait_for(
lambda: self._audio_flow_stream_id != stream_id
or self._audio_credits > 0 or self._audio_flow_error is not None
)
if self._audio_flow_stream_id != stream_id:
raise WatcheRobotError("audio stream was replaced")
if self._audio_flow_error is not None:
raise WatcheRobotError(f"audio stream failed: {self._audio_flow_error}")
self._audio_credits -= 1
class DeviceContext(ApplicationContext):
@property
def connection_lost(self):
return self._transport.closed
@classmethod
def from_environment(cls):
required = (
"WATCHER_APP_ID", "WATCHER_APP_RUN_CREDENTIAL",
"WATCHER_APP_DESKTOP_WS_URL", "WATCHER_APP_DEVICE_WS_URL",
)
environment = {name: os.environ.get(name, "").strip() for name in required}
missing = [name for name, value in environment.items() if not value]
if missing:
raise ApplicationEnvironmentError(
"Application environment is incomplete: " + ", ".join(sorted(missing))
)
return cls(app_id=environment["WATCHER_APP_ID"], transport=DeviceTransport())
|