Skip to content

Instantly share code, notes, and snippets.

@hirokiky
Last active January 24, 2022 05:08
Show Gist options
  • Save hirokiky/f4dae78b6d637f078e1c to your computer and use it in GitHub Desktop.
Save hirokiky/f4dae78b6d637f078e1c to your computer and use it in GitHub Desktop.
Periodic calling with asyncio
import asyncio
import logging
import time
import psutil
logger = logging.getLogger(__name__)
@asyncio.coroutine
def pace_maker(tasks):
for task in tasks:
task()
o = time.time()
logger.debug("Called at", o)
yield from asyncio.sleep(1-(o-int(o)))
yield from pace_maker(tasks)
def cpu_percent():
p = psutil.cpu_percent()
print("CPU usage", p, "%:", '*'*int(p // 10))
def main():
asyncio.Task(pace_maker([cpu_percent]))
loop = asyncio.get_event_loop()
try:
loop.run_forever()
finally:
loop.close()
if __name__ == "__main__":
main()
@mnowotnik
Copy link

In line 19 change yield from to asyncio.async(...)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment