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
|
import subprocess
from datetime import datetime
from config import config
import requests
from multiprocessing import Process
def tts_command(text):
if config["engine"]["tts"] == "espeak":
return ["espeak", text]
elif config["engine"]["tts"] == "festival":
filename = "/tmp/phone_input.txt"
with open(filename, "w") as f:
f.write(text)
return ["festival", "--tts", filename]
class Event():
def __init__(self, phone, args):
self.args = args
self.phone = phone
def get_name(self):
return "generic event"
def get_text(self):
return "event"
def run(self):
self.phone.run_until_hangup(tts_command(self.get_text()))
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.phone.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}"
commands = [tts_command(text), ["aplay", "/etc/phone/sound/hold.wav"]]
self.phone.run_many_until_hangup(commands)
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):
self.phone.run_until_hangup(tts_command("Please record a message."))
process = subprocess.Popen(["arecord", "/tmp/out.wav"])
self.phone.wait_until_hung_up()
process.terminate()
return "recording done"
class PlayEvent(Event):
def get_name(self):
return "play"
def run(self):
self.phone.run_until_hangup(["aplay", "/tmp/out.wav"])
class WavEvent(Event):
def get_name(self):
return self.args[0]
def run(self):
self.phone.run_until_hangup(["aplay", self.args[0]])
|
