You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Kivy+asyncio实现BLE通信:Windows正常,Linux/Android异常及事件循环警告

在Kivy应用中整合asyncio与BLE通信的事件循环解决方案

核心问题分析

你遇到的DeprecationWarning(无当前事件循环),本质是Kivy的GUI事件循环与asyncio的异步事件循环在Linux/Android平台上未正确绑定。Windows系统下asyncio默认会自动创建并绑定当前循环,但Linux/Android的环境初始化逻辑不同,导致asyncio无法找到有效运行的事件循环。

分步解决方案

1. 启动时初始化并绑定asyncio事件循环

在App的build方法中,主动检查并创建asyncio事件循环,确保其成为当前循环,从根源避免警告:

import asyncio
from kivy.app import App

class BLEApp(App):
    def build(self):
        # 初始化asyncio事件循环
        try:
            self.loop = asyncio.get_running_loop()
        except RuntimeError:
            self.loop = asyncio.new_event_loop()
            asyncio.set_event_loop(self.loop)
        
        # 后续GUI组件初始化...

2. 使用Kivy原生的协程调度方法

放弃直接调用asyncio.ensure_future或create_task,改用Kivy提供的Clock.schedule_async方法调度协程。该方法会自动协调整个应用的事件循环,确保协程在正确上下文执行,跨平台兼容性更好:

from kivy.clock import Clock

def on_send_button_press(self, instance):
    # 直接调度send_message_async协程
    Clock.schedule_async(self.send_message_async)

3. 跨平台权限与环境适配

  • Linux:确保当前用户属于bluetooth组(执行sudo usermod -aG bluetooth $USER后重启),并开启蓝牙服务(sudo systemctl start bluetooth)。
  • Android:在buildozer打包配置文件buildozer.spec中添加必要权限:
    android.permissions = BLUETOOTH, BLUETOOTH_ADMIN, ACCESS_FINE_LOCATION, BLUETOOTH_SCAN, BLUETOOTH_CONNECT
    
    (权限需根据目标Android版本调整,Android 12+需单独申请BLUETOOTH_SCAN和BLUETOOTH_CONNECT)

完整示例代码

import asyncio
from kivy.app import App
from kivy.uix.button import Button
from kivy.clock import Clock

class MicrobitBLEApp(App):
    def build(self):
        # 初始化asyncio事件循环
        try:
            self.loop = asyncio.get_running_loop()
        except RuntimeError:
            self.loop = asyncio.new_event_loop()
            asyncio.set_event_loop(self.loop)
        
        self.send_btn = Button(text="Send to micro:bit")
        self.send_btn.bind(on_press=self.on_send_press)
        return self.send_btn
    
    def on_send_press(self, instance):
        Clock.schedule_async(self.send_message_async)
    
    async def send_message_async(self):
        # 替换为你的实际BLE通信逻辑(示例用bleak库)
        try:
            from bleak import BleakClient
            MICROBIT_MAC = "你的micro:bit蓝牙MAC地址"
            UART_CHAR_UUID = "6E400002-B5A3-F393-E0A9-E50E24DCCA9E"
            
            async with BleakClient(MICROBIT_MAC) as client:
                await client.write_gatt_char(UART_CHAR_UUID, b"Hello from Kivy!")
            print("消息发送成功")
        except Exception as e:
            print(f"发送失败: {str(e)}")

if __name__ == "__main__":
    MicrobitBLEApp().run()

关键说明

  • Clock.schedule_async是Kivy官方推荐的异步任务调度方式,会自动处理与GUI循环的同步,避免线程冲突。
  • 初始化事件循环时,优先尝试获取已运行的循环,避免重复创建导致的资源冲突。
  • BLE通信逻辑必须完全放在async协程中,禁止在主线程执行阻塞操作,否则会导致GUI卡顿。

内容的提问来源于stack exchange,提问作者123

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.17 19:22:49