-
Notifications
You must be signed in to change notification settings - Fork 1
/
alarm_manager.py
131 lines (116 loc) · 4.34 KB
/
alarm_manager.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
import os.path
import pickle
import signal
from collections import defaultdict
from datetime import date, datetime, time
from operator import attrgetter
from typing import DefaultDict, List, Optional, Set, Union
from loguru import logger
from prettytable import PrettyTable
from alarmix.schema import RequestAction, TimeMessageSocket, When
from alarmix.utils import DeltaAlarm, add_delta_to_alarms, calculate_auto_time
class AlarmManager:
"""
AlarmManager manipulates your alarms
"""
def __init__(self, dump_file: str):
self.dump_file = dump_file
self.alarm_pid: Optional[int] = None
self.alarms: DefaultDict[str, Set[Union[time, datetime]]] = defaultdict(set)
def process_message(self, msg: TimeMessageSocket) -> str:
"""
Update alarms by TimeMessageSocket action.
Delete|Add|List|Stop.
"""
message = "Something happened"
if msg.action == RequestAction.delete:
self.del_alarm(msg.time, msg.when)
self.dump_alarms()
message = "Successfully deleted"
elif msg.action == RequestAction.add:
self.add_alarm(msg.time, msg.when)
self.dump_alarms()
message = "Successfully added"
elif msg.action == RequestAction.list:
message = self.list_formatted()
elif msg.action == RequestAction.stop:
message = self.stop_alarm()
return message
def list_formatted(self) -> str:
"""
Will return string containing table. Such as:
+------------+----------------+
| alarm time | remaining time |
+------------+----------------+
| 10:0 | 11:36:06 |
+------------+----------------+
"""
table = PrettyTable(field_names=["alarm time", "remaining time"])
alarms = self.list_alarms()
for alarm in alarms:
table.add_row(
[
f"{alarm.time.hour}:{alarm.time.minute}",
str(alarm.delta).split(".")[0],
]
)
return table.get_string()
def add_alarm(self, event_time: time, when: When) -> None:
logger.debug(f"Adding {event_time}")
target = event_time
if when == When.auto:
target = calculate_auto_time(event_time)
self.alarms[when.value].add(target)
def del_alarm(self, event_time: time, when: When) -> None:
"""
Delete alarm from queue
"""
logger.debug(f"Trying delete {event_time}")
target = event_time
if when == When.auto:
target = calculate_auto_time(event_time)
self.alarms[when.value].discard(target)
def list_alarms(self) -> List[DeltaAlarm]:
"""
List alarms sorted by time
"""
needed_keys = [When.everyday.value]
if date.today().weekday() < 5:
needed_keys.append(When.weekdays.value)
else:
needed_keys.append(When.weekends.value)
alarms = set()
for key in needed_keys:
alarms.update(self.alarms[key])
for alarm in self.alarms[When.auto.value]:
if date.today() == alarm.date(): # type: ignore
alarms.add(alarm.time()) # type: ignore
delta_alarms = add_delta_to_alarms(alarms)
return list(sorted(delta_alarms, key=attrgetter("delta")))
def stop_alarm(self) -> str:
"""
Stop alarm by pid.
Buzzer thread updates alarm_pid variable.
"""
if self.alarm_pid:
os.kill(self.alarm_pid, signal.SIGTERM)
self.alarm_pid = None
return "Alarm stopped"
return "Alarm isn't running"
def cleanup(self) -> None:
"""
Remove all outdated auto calculated alarms.
"""
correct_alarms = set()
for alarm_time in self.alarms[When.auto.value]:
if alarm_time >= datetime.now(): # type: ignore
correct_alarms.add(alarm_time)
self.alarms[When.auto.value] = correct_alarms
def dump_alarms(self) -> None:
with open(self.dump_file, "wb") as file:
pickle.dump(self.alarms, file)
def load_alarms(self) -> None:
if not os.path.exists(self.dump_file):
return None
with open(self.dump_file, "rb") as file:
self.alarms = pickle.load(file)