python - 如何在 python 中创建自定义计时器触发的 iotedge 函数？

Question

我的要求是每 5 分钟在我的设备上收集一次数据。这类似于每 5 分钟触发一次的函数应用程序。我的设备资源有限，我希望用 Python 做到这一点。

我尝试过类似的方法，它会一直休眠到时钟是 5 的倍数。

from azure.iot.device.aio import IoTHubModuleClient
from azure.iot.device import Message
from six.moves import input
import asyncio
from datetime import datetime, timedelta
async def main():
    sample_interval = 5
    executionTime = datetime.utcnow()
    try:
        nextSample = executionTime + timedelta(minutes=5)
        secondsUntilSampleInterval = (nextSample - datetime.utcnow()).seconds

        while True:
            print("module is connected: " + str(module_client.connected))
            if not module_client.connected:
                await module_client.connect()
            
            await asyncio.gather(asyncio.sleep(secondsUntilSampleInterval), GatherData())
            secondsUntilSampleInterval = sample_interval*60
    except Exception as e:
        print("Unexpected error %s " % e)
        raise

然而，这不符合我的要求，因为它会随着时间的推移而漂移。

我希望在时钟为 10:00、10:05、10:10 时触发 GatherData 函数...

编辑：我忘了提到 GatherData 函数是异步的。

Answer 1

我们可以使用线程..

导入线程并使用 Timer 函数每 5 秒执行一次我们的函数。下面是我的测试样本：

import threading

def runfor5sec():
  threading.Timer(5.0, runfor5sec).start() #5 is sec, convert 5 min into sec and replace.
  print("Completed 5 Sec")

runfor5sec()

在你的情况下 GatherData()

threading.Timer(5.0, GatherData).start() #5 is sec, convert 5 min into sec and 

输出：