Spaces:

left0ver
/

WatcherRobot-com.left0ver.agentcade

Running

WatcherRobot-com.left0ver.agentcade / src /agentcade /output.py

Publish com.left0ver.agentcade 0.2.0

21e8d36 verified 9 days ago

9.81 kB

import asyncio
import logging

from watcherobot.errors import CommandError
from watcherobot.job import Job

from agentcade.audio_stream import PcmAudioPlayer, sdk_command
from agentcade.eyes import STANDBY_EYES
from agentcade.robot_calls import describe_call
from agentcade.speech import QwenSpeech
from observability.events import bus, describe, failure, scope

logger = logging.getLogger(__name__)

SYNTHESIS_ATTEMPTS = 3
# 游戏的 SDK 调用返回 Job（动画、动作、音效）时，等它做完再执行下一条，最多等这么久；
# 超时就先执行后面的，Job 在机器人上继续。灯效可以一直循环，不等它。
JOB_WAIT_SECONDS = 10
NO_WAIT = {"lights"}
# 这些模块的调用会发出声音或者转动电机，执行时暂停收音；麦克风这时重新打开，机器人还会取消正在做的动作。
NOISY = {"motion", "audio", "behavior"}
# 连续几次有声音或动作的调用之间不恢复收音，免得反复开关麦克风。
HUSH_GRACE_SECONDS = 0.5
# 这些模块会改变屏幕，一局结束后屏幕要回到待机。
SCREEN = {"expression_runtime", "expressions", "animation", "behavior", "works"}
# 头部的正前方（水平角, 俯仰角），一局结束后转回这里。
NEUTRAL = (90, 115)


class Hush:
    """游戏做动作、放音效时暂停收音；连续的几次调用合成一次暂停。"""

    def __init__(self, microphone):
        self.input = microphone
        self.held = False
        self.timer: asyncio.Task | None = None

    async def hold(self):
        if self.timer is not None:
            self.timer.cancel()
            self.timer = None
        if not self.held:
            await self.input.pause(owner=self)
            self.held = True

    def release(self):
        if self.held and self.timer is None:
            self.timer = asyncio.create_task(self.release_later())

    async def release_later(self):
        await asyncio.sleep(HUSH_GRACE_SECONDS)
        self.timer = None
        self.resume()

    def resume(self):
        if self.held:
            self.held = False
            self.input.resume(owner=self)

    def close(self):
        if self.timer is not None:
            self.timer.cancel()
            self.timer = None
        self.resume()


class OutputCoordinator:
    """按顺序执行主 Agent 的输出：说话、游戏的 SDK 调用，以及一局结束后的复位。"""

    def __init__(self, app, settings, microphone):
        self.app, self.robot, self.input = app, app.robot, microphone
        self.speech = QwenSpeech(settings)
        self.last_text = ""
        self.audio = PcmAudioPlayer(self.robot.audio)
        self.speaking = False
        self.output_lock = asyncio.Lock()
        self.hush = Hush(microphone)
        # 这一局游戏调用过的 SDK 模块，结束时只复位这些。
        self.touched: set[str] = set()

    async def execute(self, output):
        async with self.output_lock:
            try:
                performed = await self.present(output)
            except BaseException as error:
                bus.emit("device.output", {"output": output, **failure(error)})
                raise
            else:
                if performed is not False:
                    bus.emit("device.output", {"output": output, "completed": True})

    async def clear(self):
        """屏幕回到待机的眼睛。"""
        await sdk_command(self.robot.expression_runtime.start, "standby", **STANDBY_EYES)

    async def perform(self, call):
        """执行游戏的一次 SDK 调用。参数由 SDK 自己检查；返回 Job 的等它做完。"""
        domain, method = call.call.split(".")
        function = getattr(getattr(self.robot, domain), method)
        self.touched.add(domain)
        noisy = domain in NOISY
        if noisy:
            await self.hush.hold()
        try:
            for attempt in range(3):
                try:
                    result = await sdk_command(function, *call.args, **call.kwargs)
                    break
                except CommandError as error:
                    # 固件偶尔对连续的表情更新返回 invalid_argument，稍等重试。
                    if (error.message_type != "ctrl.expression.runtime.update"
                            or error.reason != "invalid_argument" or attempt == 2):
                        raise
                    logger.warning("表情更新返回 invalid_argument，0.2 秒后重试（%s/2）", attempt + 1)
                    await asyncio.sleep(0.2)
            if isinstance(result, Job) and domain not in NO_WAIT:
                try:
                    await sdk_command(result.wait, JOB_WAIT_SECONDS)
                except TimeoutError:
                    logger.info("%s 超过 %s 秒还没做完，先执行后面的", call.call, JOB_WAIT_SECONDS)
        finally:
            if noisy:
                self.hush.release()

    async def reset(self):
        """一局结束后把游戏用过的设备复位：停掉动画和行为、关灯、头转回正前方、屏幕回到待机。
        游戏没有调用过 SDK 时什么也不做，返回 False。
        """
        touched, self.touched = self.touched, set()
        if not touched:
            return False
        steps = []
        if "animation" in touched:
            steps.append(("停止动画", self.robot.animation.stop, {}, False))
        if "behavior" in touched:
            steps.append(("停止行为", self.robot.behavior.stop, {}, False))
        if "lights" in touched:
            steps.append(("关灯", self.robot.lights.off, {}, False))
        if "motion" in touched:
            pan, tilt = NEUTRAL
            steps.append(("头转回正前方", self.robot.motion.move_to,
                          {"pan_deg": pan, "tilt_deg": tilt, "duration_ms": 400}, True))
        # 复位紧跟在最后一句话后面，说完收音会重新打开，转头时要暂停收音。
        for name, function, arguments, noisy in steps:
            if noisy:
                await self.hush.hold()
            try:
                result = await sdk_command(function, **arguments)
                if isinstance(result, Job):
                    await sdk_command(result.wait, JOB_WAIT_SECONDS)
            except Exception as error:
                # 复位做不完只影响外观，接着复位其他的。
                logger.warning("一局结束后%s失败：%s", name, describe(error), exc_info=True)
            finally:
                if noisy:
                    self.hush.release()
        if touched & SCREEN:
            await self.clear()

    async def synthesize(self, text):
        for attempt in range(1, SYNTHESIS_ATTEMPTS + 1):
            try:
                return await self.speech.synthesize(text)
            except ValueError:
                raise
            except Exception as error:
                # 语音服务偶尔连接失败或超时，重试仍然失败再放弃这一条。
                if attempt == SYNTHESIS_ATTEMPTS:
                    raise
                logger.warning("语音合成失败，重试（%s/%s）：%s",
                               attempt, SYNTHESIS_ATTEMPTS - 1, describe(error))
                await asyncio.sleep(0.5 * attempt)

    async def present(self, output):
        if output.call is not None:
            return await self.perform(output.call)
        if output.reset:
            return await self.reset()
        if not output.text:
            return False
        self.speaking = True
        await self.input.pause(owner=self)
        try:
            audio = await self.synthesize(output.text)
            self.last_text = output.text
            self.app.logger.info("机器人：%s", output.text)
            metrics = await self.audio.play(audio, on_start=lambda seconds: self.started(output, seconds))
            self.app.logger.info(
                "语音播放：任务 %s，启动 %.2f 秒，完成 %.2f 秒，PCM %s 字节",
                metrics.playback_id, metrics.started_seconds, metrics.completed_seconds,
                metrics.pcm_bytes,
            )
        finally:
            self.speaking = False
            self.input.resume(owner=self)

    async def started(self, output, seconds):
        # 对话页面按这段语音的时长把文字逐步显示出来，和机器人说话的进度一致。
        bus.emit("speech.start", {"text": output.text, "duration_ms": round(seconds * 1000)})

    async def run(self, agent):
        while True:
            request = await agent.outputs.get()
            try:
                completed = []
                with scope(**request.trace_context):
                    for output in request.outputs:
                        try:
                            await self.execute(output)
                        except Exception as error:
                            # 一条输出失败只跳过这一条，后面的播报照常进行。
                            label = (output.text[:30] if output.text
                                     else describe_call(output.call) if output.call else "复位")
                            logger.warning("输出失败，已跳过“%s”：%s", label, describe(error), exc_info=True)
                            continue
                        completed.append(output)
                request.outputs = completed
                await agent.output_done(request)
            finally:
                if not request.completed.done():
                    request.completed.cancel()
                agent.outputs.task_done()

    async def close(self):
        self.hush.close()
        # 退出时还在玩的那一局不会再有结束的回复，直接复位。
        async with self.output_lock:
            await self.reset()

	import asyncio
	import logging

	from watcherobot.errors import CommandError
	from watcherobot.job import Job

	from agentcade.audio_stream import PcmAudioPlayer, sdk_command
	from agentcade.eyes import STANDBY_EYES
	from agentcade.robot_calls import describe_call
	from agentcade.speech import QwenSpeech
	from observability.events import bus, describe, failure, scope

	logger = logging.getLogger(__name__)

	SYNTHESIS_ATTEMPTS = 3
	# 游戏的 SDK 调用返回 Job（动画、动作、音效）时，等它做完再执行下一条，最多等这么久；
	# 超时就先执行后面的，Job 在机器人上继续。灯效可以一直循环，不等它。
	JOB_WAIT_SECONDS = 10
	NO_WAIT = {"lights"}
	# 这些模块的调用会发出声音或者转动电机，执行时暂停收音；麦克风这时重新打开，机器人还会取消正在做的动作。
	NOISY = {"motion", "audio", "behavior"}
	# 连续几次有声音或动作的调用之间不恢复收音，免得反复开关麦克风。
	HUSH_GRACE_SECONDS = 0.5
	# 这些模块会改变屏幕，一局结束后屏幕要回到待机。
	SCREEN = {"expression_runtime", "expressions", "animation", "behavior", "works"}
	# 头部的正前方（水平角, 俯仰角），一局结束后转回这里。
	NEUTRAL = (90, 115)


	class Hush:
	"""游戏做动作、放音效时暂停收音；连续的几次调用合成一次暂停。"""

	def __init__(self, microphone):
	self.input = microphone
	self.held = False
	self.timer: asyncio.Task \| None = None

	async def hold(self):
	if self.timer is not None:
	self.timer.cancel()
	self.timer = None
	if not self.held:
	await self.input.pause(owner=self)
	self.held = True

	def release(self):
	if self.held and self.timer is None:
	self.timer = asyncio.create_task(self.release_later())

	async def release_later(self):
	await asyncio.sleep(HUSH_GRACE_SECONDS)
	self.timer = None
	self.resume()

	def resume(self):
	if self.held:
	self.held = False
	self.input.resume(owner=self)

	def close(self):
	if self.timer is not None:
	self.timer.cancel()
	self.timer = None
	self.resume()


	class OutputCoordinator:
	"""按顺序执行主 Agent 的输出：说话、游戏的 SDK 调用，以及一局结束后的复位。"""

	def __init__(self, app, settings, microphone):
	self.app, self.robot, self.input = app, app.robot, microphone
	self.speech = QwenSpeech(settings)
	self.last_text = ""
	self.audio = PcmAudioPlayer(self.robot.audio)
	self.speaking = False
	self.output_lock = asyncio.Lock()
	self.hush = Hush(microphone)
	# 这一局游戏调用过的 SDK 模块，结束时只复位这些。
	self.touched: set[str] = set()

	async def execute(self, output):
	async with self.output_lock:
	try:
	performed = await self.present(output)
	except BaseException as error:
	bus.emit("device.output", {"output": output, **failure(error)})
	raise
	else:
	if performed is not False:
	bus.emit("device.output", {"output": output, "completed": True})

	async def clear(self):
	"""屏幕回到待机的眼睛。"""
	await sdk_command(self.robot.expression_runtime.start, "standby", **STANDBY_EYES)

	async def perform(self, call):
	"""执行游戏的一次 SDK 调用。参数由 SDK 自己检查；返回 Job 的等它做完。"""
	domain, method = call.call.split(".")
	function = getattr(getattr(self.robot, domain), method)
	self.touched.add(domain)
	noisy = domain in NOISY
	if noisy:
	await self.hush.hold()
	try:
	for attempt in range(3):
	try:
	result = await sdk_command(function, call.args, *call.kwargs)
	break
	except CommandError as error:
	# 固件偶尔对连续的表情更新返回 invalid_argument，稍等重试。
	if (error.message_type != "ctrl.expression.runtime.update"
	or error.reason != "invalid_argument" or attempt == 2):
	raise
	logger.warning("表情更新返回 invalid_argument，0.2 秒后重试（%s/2）", attempt + 1)
	await asyncio.sleep(0.2)
	if isinstance(result, Job) and domain not in NO_WAIT:
	try:
	await sdk_command(result.wait, JOB_WAIT_SECONDS)
	except TimeoutError:
	logger.info("%s 超过 %s 秒还没做完，先执行后面的", call.call, JOB_WAIT_SECONDS)
	finally:
	if noisy:
	self.hush.release()

	async def reset(self):
	"""一局结束后把游戏用过的设备复位：停掉动画和行为、关灯、头转回正前方、屏幕回到待机。

	游戏没有调用过 SDK 时什么也不做，返回 False。
	"""
	touched, self.touched = self.touched, set()
	if not touched:
	return False
	steps = []
	if "animation" in touched:
	steps.append(("停止动画", self.robot.animation.stop, {}, False))
	if "behavior" in touched:
	steps.append(("停止行为", self.robot.behavior.stop, {}, False))
	if "lights" in touched:
	steps.append(("关灯", self.robot.lights.off, {}, False))
	if "motion" in touched:
	pan, tilt = NEUTRAL
	steps.append(("头转回正前方", self.robot.motion.move_to,
	{"pan_deg": pan, "tilt_deg": tilt, "duration_ms": 400}, True))
	# 复位紧跟在最后一句话后面，说完收音会重新打开，转头时要暂停收音。
	for name, function, arguments, noisy in steps:
	if noisy:
	await self.hush.hold()
	try:
	result = await sdk_command(function, **arguments)
	if isinstance(result, Job):
	await sdk_command(result.wait, JOB_WAIT_SECONDS)
	except Exception as error:
	# 复位做不完只影响外观，接着复位其他的。
	logger.warning("一局结束后%s失败：%s", name, describe(error), exc_info=True)
	finally:
	if noisy:
	self.hush.release()
	if touched & SCREEN:
	await self.clear()

	async def synthesize(self, text):
	for attempt in range(1, SYNTHESIS_ATTEMPTS + 1):
	try:
	return await self.speech.synthesize(text)
	except ValueError:
	raise
	except Exception as error:
	# 语音服务偶尔连接失败或超时，重试仍然失败再放弃这一条。
	if attempt == SYNTHESIS_ATTEMPTS:
	raise
	logger.warning("语音合成失败，重试（%s/%s）：%s",
	attempt, SYNTHESIS_ATTEMPTS - 1, describe(error))
	await asyncio.sleep(0.5 * attempt)

	async def present(self, output):
	if output.call is not None:
	return await self.perform(output.call)
	if output.reset:
	return await self.reset()
	if not output.text:
	return False
	self.speaking = True
	await self.input.pause(owner=self)
	try:
	audio = await self.synthesize(output.text)
	self.last_text = output.text
	self.app.logger.info("机器人：%s", output.text)
	metrics = await self.audio.play(audio, on_start=lambda seconds: self.started(output, seconds))
	self.app.logger.info(
	"语音播放：任务 %s，启动 %.2f 秒，完成 %.2f 秒，PCM %s 字节",
	metrics.playback_id, metrics.started_seconds, metrics.completed_seconds,
	metrics.pcm_bytes,
	)
	finally:
	self.speaking = False
	self.input.resume(owner=self)

	async def started(self, output, seconds):
	# 对话页面按这段语音的时长把文字逐步显示出来，和机器人说话的进度一致。
	bus.emit("speech.start", {"text": output.text, "duration_ms": round(seconds * 1000)})

	async def run(self, agent):
	while True:
	request = await agent.outputs.get()
	try:
	completed = []
	with scope(**request.trace_context):
	for output in request.outputs:
	try:
	await self.execute(output)
	except Exception as error:
	# 一条输出失败只跳过这一条，后面的播报照常进行。
	label = (output.text[:30] if output.text
	else describe_call(output.call) if output.call else "复位")
	logger.warning("输出失败，已跳过“%s”：%s", label, describe(error), exc_info=True)
	continue
	completed.append(output)
	request.outputs = completed
	await agent.output_done(request)
	finally:
	if not request.completed.done():
	request.completed.cancel()
	agent.outputs.task_done()

	async def close(self):
	self.hush.close()
	# 退出时还在玩的那一局不会再有结束的回复，直接复位。
	async with self.output_lock:
	await self.reset()