File size: 2,597 Bytes
9f5687a 7c651d5 9f5687a 7c651d5 9f5687a 7c651d5 9f5687a 7c651d5 9f5687a 7c651d5 9f5687a 7c651d5 9f5687a 7c651d5 | 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 | """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())
|