Python调用LightsOn/LightsOff函数后无报错停止运行问题排查

问题根源 #

程序在调用LightsOn()或LightsOff()后静默终止，核心原因有两个：

1. asyncio.run()执行完毕会自动关闭事件循环，而kasa库的SmartStrip实例依赖事件循环运行，后续再调用设备方法时会因无可用事件循环崩溃（无报错是因为未捕获该异常）
2. 混用同步库pyHS100与异步库kasa，两者IO模型冲突，导致设备连接异常

修复后的完整代码 #

import requests
import asyncio
from kasa import SmartStrip, Discover

url = 'https://awesomebits.online/SmartHome/controlLights/light_status.txt'
stat = ''
previous_content = None
dev = None
bedroom_lights = None

async def init_devices():
    global dev, bedroom_lights
    # 初始化智能插排设备
    dev = SmartStrip("192.168.7.105")
    await dev.update()  # 必须先更新设备状态，否则无法正常控制
    
    # 用kasa异步发现设备，替代同步的pyHS100
    discovered_devices = await Discover.discover()
    bedroom_lights = next((dev for dev in discovered_devices.values() if dev.alias == "Bens Bedroom lights"), None)
    
    if bedroom_lights is None:
        print("未找到Bens Bedroom lights设备。")
    else:
        await bedroom_lights.update()

async def check_and_control_lights():
    global previous_content, stat
    while True:
        # 获取卧室灯当前状态
        bedroom_lights_status = bedroom_lights.is_on if bedroom_lights else None
        
        if bedroom_lights_status is not None:
            print("灯已开启" if bedroom_lights_status else "灯已关闭")
            stat = 1 if bedroom_lights_status else 0
        else:
            print("未找到Bens Bedroom lights设备。")
        
        # 获取远程控制指令
        try:
            current_content = requests.get(url).text.strip().lower()
            print(f"远程指令: {current_content}")
        except Exception as e:
            print(f"获取远程指令失败: {e}")
            await asyncio.sleep(0.5)
            continue
        
        # 指令变更时执行操作
        if current_content != previous_content:
            if current_content == 'on' and stat == 0:
                print("正在打开灯光...")
                await dev.turn_on()
                print("灯光已开启")
            elif current_content == 'off' and stat == 1:
                print("正在关闭灯光...")
                await dev.turn_off()
                print("灯光已关闭")
            else:
                print("无效指令或状态无需变更")
            
            previous_content = current_content
        
        await asyncio.sleep(0.5)  # 异步休眠，不阻塞事件循环

async def main():
    await init_devices()
    await check_and_control_lights()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception as e:
        print(f"程序异常终止: {e}")
        import traceback
        traceback.print_exc()

关键修改说明 #

• 全程异步化：整个程序基于asyncio事件循环运行，避免同步/异步代码混用的冲突
• 移除pyHS100库：kasa自带异步设备发现功能，无需依赖同步库
• 设备初始化必须调用update()：kasa设备需要先同步状态才能正常控制
• 用asyncio.sleep替代time.sleep：time.sleep会阻塞整个程序，asyncio.sleep只挂起当前任务，保证其他操作正常运行
• 全局异常捕获：捕获所有未处理异常，避免程序静默终止，方便排查问题

内容的提问来源于stack exchange，提问作者john deere