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
|
import subprocess
from datetime import datetime
from config import config
import requests
from multiprocessing import Process
def run_until_hangup(command, phone_hung_up):
'''
returns true if process was finished
'''
process = subprocess.Popen(command)
while process.poll() is None:
if phone_hung_up.wait(timeout=0.1):
process.terminate()
return False
return True
class Event():
def __init__(self, dial_manager, phone_held, phone_hung_up, args):
self.dial_manager = dial_manager
self.phone_held = phone_held
self.phone_hung_up = phone_hung_up
self.args = args
def get_name(self):
return "generic event"
def get_text(self):
return "event"
def run(self):
run_until_hangup(["espeak", self.get_text()], self.phone_hung_up)
class FortuneEvent(Event):
def get_name(self):
return "fortune"
def get_text(self):
res = subprocess.run(["fortune", "-s"], stdout=subprocess.PIPE)
return res.stdout.decode("utf-8")
class DirectoryEvent(Event):
def get_name(self):
return "directory"
def get_text(self):
def recursive_build(prefix, sequences):
sequence_list = []
for key in sequences.keys():
sequence = prefix+" "+str(key)
if isinstance(sequences[key], dict):
sequence_list += recursive_build(sequence, sequences[key])
else:
sequence_list.append((sequence, sequences[key]))
return sequence_list
sequences = self.dial_manager.sequences
text = ""
for sequence, event in recursive_build("", sequences):
text += f"For {event.get_name()} dial {sequence}. "
return text
class OperatorEvent(Event):
def get_name(self):
return "operator"
def run(self):
loop = True
while loop:
now = datetime.now()
nearest_quarter = ((now.minute // 15 + 1) * 15) % 60
next_hour = now.hour % 12 if nearest_quarter != 0 else (now.hour + 1) % 12
next_hour = next_hour if next_hour != 0 else 12
text = f"In {nearest_quarter - now.minute} minutes it will be {next_hour} {nearest_quarter}"
loop = run_until_hangup(["espeak", text], self.phone_hung_up)
if loop:
loop = run_until_hangup(["aplay", "/etc/phone/sound/hold.wav"], self.phone_hung_up)
class WeatherEvent(Event):
def get_name(self):
return "weather"
def get_text(self):
key = config["openweathermap"]["key"]
lat = config["openweathermap"]["lat"]
lon = config["openweathermap"]["lon"]
location = config["openweathermap"]["location"]
units = config["openweathermap"]["units"]
onecall_url = f"https://api.openweathermap.org/data/2.5/onecall?lat={lat}&lon={lon}&appid={key}&units={units}"
onecall = requests.get(onecall_url).json()
current_temp = onecall["current"]["temp"]
current_desc = onecall["current"]["weather"][0]["description"]
weather = f"In {location}, it is {current_temp} degrees with {current_desc}. "
high_temp = onecall["daily"][0]["temp"]["max"]
pop = onecall["daily"][0]["pop"] * 100
forecast = f"Today there is a high of {high_temp} with a {pop} percent chance of percipitation."
return f"{weather}{forecast}"
class RecordEvent(Event):
def get_name(self):
return "recorder"
def run(self):
run_until_hangup(["espeak", "Please record a message."], self.phone_hung_up)
process = subprocess.Popen(["arecord", "/tmp/out.wav"])
self.phone_hung_up.wait()
process.terminate()
return "recording done"
class PlayEvent(Event):
def get_name(self):
return "play"
def run(self):
run_until_hangup(["aplay", "/tmp/out.wav"], self.phone_hung_up)
class WavEvent(Event):
def get_name(self):
return self.args[0]
def run(self):
run_until_hangup(["aplay", self.args[0]], self.phone_hung_up)
|
