树莓派Python asyncio定时控制继电器 非阻塞运行其他任务方法

• 技术栈：Python + asyncio，程序部署在连接继电器的树莓派上，用于实现智能充电控制
• 预期功能：按预设时间控制继电器通断，例如下午4点闭合继电器开启充电，下午5点断开继电器结束充电
• 现存问题：
  • 充电流程运行期间，需要保证程序其他任务可以正常并发执行
  • 需要实现固定周期（如每分钟1次）的时间检测逻辑，准确识别停止/启动充电的触发时机
• 初始实现思路存在缺陷，参考代码如下：

import datetime

startCharging = datetime.datetime(year=2022, month=7, day=7, hour=16, minute=0)
stopCharging = datetime.datetime(year=2022, month=7, day=7, hour=17, minute=0)

if(datetime.datetime.now()==startCharging):
    # Start Charging
else:
    # Stop Charging 

核心需求：让时间检测逻辑持续后台运行，同时不阻塞程序内其他函数的正常执行。

基于asyncio的异步任务调度机制实现即可，核心是把时间检测、充电控制、其他业务逻辑都拆成独立的非阻塞协程，统一交给事件循环调度，各任务在等待IO/休眠时会自动让出执行权，不会互相阻塞。

关键实现要点 #

• 不要用精确相等判断时间：程序调度存在毫秒级延迟，datetime.datetime.now()==目标时间的逻辑几乎不可能命中，很容易错过触发点，应该判断当前时间是否跨过目标时间点，同时增加状态位避免重复触发动作
• 禁止在协程中使用同步阻塞的time.sleep()做等待，必须使用异步的asyncio.sleep()，后者不会阻塞事件循环，等待期间其他协程可以正常调度执行
• 阻塞操作做异步包装：如果继电器控制库是同步阻塞的，用asyncio.to_thread()把相关操作放到独立线程执行，避免卡住事件循环
• 检测周期按需设置：每分钟1次的检测频率足够满足分钟级定时需求，对精度要求高可以把间隔缩短到5-10秒，资源占用极低

参考实现代码 #

import asyncio
import datetime

# 充电状态标记，避免重复触发继电器
is_charging = False

async def relay_control(action: str):
    """继电器控制函数，同步库就用asyncio.to_thread包装，这里写占位逻辑"""
    print(f"{datetime.datetime.now()} 执行继电器操作: {action}")
    # 如果是同步阻塞的继电器SDK，替换成下面的写法
    # await asyncio.to_thread(sync_relaySDK.set_action, action)
    await asyncio.sleep(0.1) # 模拟操作耗时

async def time_check_task():
    """后台常驻的周期时间检测任务"""
    global is_charging
    # 这里可以改成动态生成每日的启停时间，不需要写死日期
    start_charging = datetime.datetime(year=2022, month=7, day=7, hour=16, minute=0)
    stop_charging = datetime.datetime(year=2022, month=7, day=7, hour=17, minute=0)
    
    while True:
        now = datetime.datetime.now()
        # 到启动时间且当前未在充电，触发启动
        if now >= start_charging and not is_charging:
            await relay_control("闭合")
            is_charging = True
        # 到停止时间且当前在充电，触发停止
        if now >= stop_charging and is_charging:
            await relay_control("断开")
            is_charging = False
        # 等待60秒后执行下一次检测，异步等待不会阻塞其他任务
        await asyncio.sleep(60)

async def other_business_task():
    """其他需要并行运行的业务任务示例，比如状态上报、传感器采集"""
    while True:
        print(f"{datetime.datetime.now()} 其他业务任务运行中，当前充电状态: {is_charging}")
        await asyncio.sleep(20) # 模拟业务逻辑间隔

async def main():
    # 把所有任务都注册到事件循环，并发运行
    await asyncio.gather(
        time_check_task(),
        other_business_task()
    )

if __name__ == "__main__":
    asyncio.run(main())

补充说明 #

• 如果需要实现每日循环定时，只需要在每次检测时动态生成当天的start_charging、stop_charging时间即可，不需要硬编码日期
• 协程内不要写长时间阻塞的同步代码，所有耗时IO、等待操作都要换成异步实现或者丢到线程池执行，才能保证所有任务并行不悖
• 可以根据实际精度需求调整检测间隔，哪怕把检测间隔设为1秒，对树莓派的资源占用也可以忽略不计

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