app.py

"""WatcheRobot Application entrypoint: robot control console."""

import asyncio

from ui import ControlPanel
from watcherobot.application import ApplicationContext

JOB_TIMEOUT_S = 30.0


async def main() -> None:
    loop = asyncio.get_running_loop()

    async with ApplicationContext.from_environment() as app:
        panel = ControlPanel(
            on_action=lambda aid, name: loop.create_task(play_action(app, panel, aid, name)),
            on_light=lambda color, name: loop.create_task(set_light(app, panel, color, name)),
            on_lights_off=lambda: loop.create_task(lights_off(app, panel)),
            on_stop=lambda: loop.create_task(stop_all(app, panel)),
        )
        panel.set_status(f"已连接 · 应用 {app.app_id}")
        panel.log(f"控制台就绪，应用 {app.app_id} 已连接机器人。")
        app.logger.info("Control console started: %s", app.app_id)
        try:
            while panel.is_open:
                panel.pump()
                await asyncio.sleep(0.03)
        finally:
            panel.close()


async def play_action(app, panel, action_id: str, name: str) -> None:
    """Play a robot action in a worker thread so the UI never freezes."""

    try:
        job = await asyncio.to_thread(app.robot.motion.play_action, action_id)
        panel.set_status(f"播放中：{name}")
        await asyncio.to_thread(job.wait, JOB_TIMEOUT_S)
        panel.set_status("已连接")
        panel.log(f"✓ 动作完成：{name}（{action_id}）")
    except Exception as exc:  # noqa: BLE001 - surface every failure to the user
        panel.set_status("已连接（上次操作失败）")
        panel.log(f"✗ 动作失败：{name} - {exc}")


async def set_light(app, panel, color: str, name: str) -> None:
    try:
        await asyncio.to_thread(app.robot.lights.set_color, color)
        panel.log(f"✓ 灯光已设置：{name}")
    except Exception as exc:  # noqa: BLE001
        panel.log(f"✗ 灯光设置失败：{name} - {exc}")


async def lights_off(app, panel) -> None:
    try:
        await asyncio.to_thread(app.robot.lights.off)
        panel.log("✓ 已关灯")
    except Exception as exc:  # noqa: BLE001
        panel.log(f"✗ 关灯失败 - {exc}")


async def stop_all(app, panel) -> None:
    try:
        await asyncio.to_thread(app.robot.motion.stop)
        await asyncio.to_thread(app.robot.behavior.stop)
        panel.set_status("已连接")
        panel.log("✓ 已停止所有动作")
    except Exception as exc:  # noqa: BLE001
        panel.log(f"✗ 停止失败 - {exc}")


asyncio.run(main())