Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- 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())
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement