import time
import asyncio
from datetime import datetime as DateTime
from datetime import timedelta as TimeDelta
from datetime import timezone as TimeZone
from zoneinfo import ZoneInfo
from sched import scheduler
UTC = TimeZone.utc
def timefunc():
return DateTime.now(ZoneInfo("Europe/Berlin"))
def delayfunc(delay):
if isinstance(delay, TimeDelta):
delay = delay.total_seconds()
time.sleep(delay)
def wrap_sync(func, *args1, **kwargs):
async def run(*args, **kwargs):
return func(*args, **kwargs)
def wrapped(*args2):
asyncio.create_task(run(*args1, *args2, **kwargs))
return wrapped
async def main():
task_runner = scheduler(timefunc, delayfunc)
for delay in range(5, 31, 5):
abs_time = DateTime.now(ZoneInfo("Europe/Berlin")) + TimeDelta(seconds=delay)
task_runner.enterabs(abs_time, 0, wrap_sync(print, delay, abs_time))
while task_runner.queue:
task_runner.run(False)
await asyncio.sleep(0)
if __name__ == "__main__":
asyncio.run(main())
Comments
0 B
|👍
/👎