Coverage for /home/runner/work/coxbuild/coxbuild/src/coxbuild/events/datetime.py: 62%

1from datetime import date, datetime, time, timedelta 

2 

3from . import delay, limit, occur, once, onceevent 

4 

5 

6async def attime(dt: datetime | date | time): 

7 while True: 

8 now = datetime.now() 

9 

10 if isinstance(dt, datetime): 

11 cdt = dt 

12 elif isinstance(dt, date): 

13 cdt = datetime(dt.year, dt.month, dt.day, 

14 now.hour, now.minute, now.second) 

15 elif isinstance(dt, time): 

16 cdt = datetime(now.year, now.month, now.day, 

17 dt.hour, dt.minute, dt.second) 

18 if cdt < now: 

19 cdt = cdt + timedelta(days=1) 

20 

21 if cdt.date() == now.date() and cdt.hour == now.hour and cdt.minute == now.minute and cdt.second == now.second: 

22 yield 

23 elif now < cdt: 

24 await occur(delay(cdt - now)) 

25 yield 

26 else: 

27 break 

28 

29 print(dt, cdt, datetime.now()) 

30 

31 # go to next time span 

32 

33 if isinstance(dt, time): 

34 await occur(delay(timedelta(hours=1))) 

35 elif isinstance(dt, datetime): 

36 await occur(delay(timedelta(seconds=1))) 

37 

38 

39def tomorrow(): 

40 return attime(date.today() + timedelta(days=1)) 

41 

42 

43def nextWeek(): 

44 return attime(date.today() + timedelta(days=7)) 

45 

46 

47async def weekdays(*weekdays: int): 

48 while True: 

49 today = date.today() 

50 cur = today.weekday() 

51 if cur in weekdays: 

52 yield 

53 else: 

54 done = False 

55 for i in range(1, 8): 

56 if (cur + i) % 7 in weekdays: 

57 done = True 

58 await occur(attime(today+timedelta(days=i))) 

59 yield 

60 await occur(tomorrow()) 

61 if not done: 

62 break 

63 

64monday = weekdays(0) 

65tuesday = weekdays(1) 

66wednesday = weekdays(2) 

67thursday = weekdays(3) 

68friday = weekdays(4) 

69saturday = weekdays(5) 

70sunday = weekdays(6) 

71 

72weekday = weekdays(0, 1, 2, 3, 4) 

73weekend = weekdays(5, 6)