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
132
133
134
135
136
137
138
139
140
141
142
143
|
import subprocess
import requests
import smtplib
from config import config
from email.message import EmailMessage
from datetime import datetime
from config import config
from multiprocessing import Process
from os.path import basename
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", config["engine"]["filename"]])
self.phone.wait_until_hung_up()
process.terminate()
self._send_recording()
return ""
def _send_recording(self):
msg = EmailMessage()
msg["Subject"] = "Phone Recording"
msg["From"] = config["email"]["user"]
msg["To"] = config["email"]["user"]
msg.set_content("Attached is your message.")
with open(config["engine"]["filename"], 'rb') as f:
data = f.read()
msg.add_attachment(data, maintype="audio", subtype="wav", filename="recording.wav")
s = smtplib.SMTP(config["email"]["server"], int(config["email"]["port"]))
s.starttls()
s.login(config["email"]["user"], config["email"]["pass"])
s.send_message(msg)
s.quit()
class PlayEvent(Event):
def get_name(self):
return "play"
def run(self):
self.phone.run_until_hangup(["aplay", config["engine"]["filename"]])
class WavEvent(Event):
def get_name(self):
return basename(self.args[0])
def run(self):
self.phone.run_until_hangup(["aplay", self.args[0]])
|
