Coverage for src/botadi/mokadi/gui_mokadi.py: 19%

1# -*- coding: utf-8 -*- 

2""" 

3@file 

4@brief Tkinter application for Mokadi 

5""" 

6import tkinter 

7import tkinter.ttk as tttk 

8import tkinter.scrolledtext as ScrolledText 

9import os 

10import threading 

11from queue import Queue 

12from PIL import Image, ImageTk 

13from pyquickhelper.loghelper import CustomLog, fLOG, get_password 

14from .grammars import MokadiGrammar_frParser, MokadiGrammar_frLexer, MokadiGrammar_frListener 

15from .mokadi_message import MokadiMessage 

16from .mokadi_engine import MokadiEngine 

17from .mokadi_action_conversation import MokadiActionConversation 

18from .mokadi_action_emotion import MokadiActionEmotion 

19from .mokadi_action_mail import MokadiActionMail 

20from .mokadi_action_news import MokadiActionNews 

21from .mokadi_action_slides import MokadiActionSlides 

22from .mokadi_record import play_speech, record_speech 

23from .mokadi_speak import speak 

24from .cognitive_services_helper import call_api_speech_reco 

25from .mokadi_picture import take_picture 

26from .gui_mokadi_process import start_process_listen 

27 

28 

29class ThreadSpeech(threading.Thread): 

30 

31 def __init__(self, win, subkey, fLOG): 

32 threading.Thread.__init__(self) 

33 self.win = win 

34 self.subkey = subkey 

35 self.fLOG = fLOG 

36 

37 def run(self): 

38 self.fLOG("[BingSpeech] start listening") 

39 speech = record_speech() 

40 self.fLOG("[BingSpeech] call API ", len(speech)) 

41 try: 

42 reco = call_api_speech_reco(self.subkey, memwav=speech) 

43 exc = None 

44 except Exception as e: 

45 reco = "error, Je n'ai rien entendu. Etes-vous connecté ?", 1 

46 exc = e 

47 self.fLOG("[BingSpeech] received ", len(speech)) 

48 if exc is not None: 

49 self.fLOG("[BingSpeech] exc", exc) 

50 elif "results" not in reco: 

51 reco = "error, Je n'ai rien entendu. Etes-vous sûr d'avoir parlé ?", 1 

52 else: 

53 results = reco["results"] 

54 if len(results) != 1: 

55 reco = "error, Je n'ai rien compris. Avez-vous parlé assez fort ?", 1 

56 else: 

57 res = results[0] 

58 if "lexical" not in res: 

59 reco = "error, Quelle langue avez-vous parlé ?", 1 

60 else: 

61 conf = res["confidence"] 

62 reco = res["lexical"], conf 

63 

64 self.win.queue.put_nowait(reco) 

65 self.win.event_generate("<<thread_fini>>") 

66 

67 

68class ThreadListenProcess(threading.Thread): 

69 

70 def __init__(self, win, fLOG): 

71 threading.Thread.__init__(self) 

72 self.win = win 

73 self.fLOG = fLOG 

74 

75 def run(self): 

76 self.fLOG("[Listen] start") 

77 process, parent_conn, _ = start_process_listen() 

78 self.process = process 

79 good = {'123', 'et 23', '23', 'et 223', '923', '2123', ', 923', 'l\'un des trois', 

80 '0123', 'Trois'} 

81 while True: 

82 text = parent_conn.recv() 

83 self.fLOG("[Listen]", text) 

84 if text in good: 

85 speak("Je t'écoute.") 

86 self.win.event_generate("<<Listener>>") 

87 

88 def stop_listening(self): 

89 self.process.terminate() 

90 

91 

92class ThreadListen(threading.Thread): 

93 

94 def __init__(self, win, fLOG, folder): 

95 threading.Thread.__init__(self) 

96 self.win = win 

97 self.fLOG = fLOG 

98 self.folder = folder 

99 self.filename = os.path.join(folder, "roaming_listening.wav") 

100 

101 def run(self): 

102 self.fLOG("[Listen] start") 

103 # Implementation was removed. 

104 # from ensae_teaching_cs.cspython import vocal_recognition_system 

105 good = {'123', 'et 23', '23', 'et 223', '923', '2123', ', 923', 'l\'un des trois', 

106 '0123'} 

107 self.continue_listening = True 

108 while self.continue_listening: 

109 self.fLOG("[Listen] .") 

110 sound = record_speech(RECORD_SECONDS=2) 

111 with open(self.filename, "wb") as f: 

112 f.write(sound) 

113 res = None # vocal_recognition_system(self.filename) 

114 if res is not None: 

115 text = " ".join(_[1] for _ in res) # pylint: disable=E1133 

116 self.fLOG("[Listen] .", text) 

117 if text in good: 

118 self.win.event_generate("<<Listener>>") 

119 

120 def stop_listening(self): 

121 self.continue_listening = False 

122 

123 

124class TkinterMokadi(tkinter.Frame): 

125 """ 

126 Defines a frame. 

127 """ 

128 

129 def __init__(self, parent, mokadi, vocal=False, subkey_speech=None, 

130 fLOG=fLOG, folder="."): 

131 """ 

132 @param parent a frame 

133 @param mokadi the bot @see cl MokadiEngine 

134 @param vocal speak the answer and not just display it 

135 @param subkey_speech key for the speech 

136 @param folder temporary folder 

137 @param fLOG logging function 

138 """ 

139 tkinter.Frame.__init__(self, parent) 

140 self._mokadi = mokadi 

141 self._speak = vocal 

142 self._subkey_speech = subkey_speech 

143 self.fLOG = fLOG 

144 self.queue = Queue() 

145 self._folder = folder 

146 self.initialize() 

147 

148 def initialize(self): 

149 """ 

150 Initialisation. 

151 """ 

152 self.grid() 

153 self.subframe1 = tkinter.Frame(self) 

154 self.subframe2 = tkinter.Frame(self) 

155 self.subframe1.grid(column=0, row=0) 

156 self.subframe2.grid(column=1, row=0) 

157 

158 self.respond = tttk.Button( 

159 self.subframe1, text='Demander', command=self.get_response) 

160 self.respond.grid(column=1, row=0, sticky='nesw', padx=3, pady=3) 

161 self.respond.config(width=5) 

162 

163 self.listen = tttk.Button( 

164 self.subframe1, text='Ecouter', command=self.start_speech_listening) 

165 self.listen.grid(column=1, row=1, sticky='nesw', padx=3, pady=3) 

166 self.listen.config(width=5) 

167 

168 self.usr_input = tttk.Entry(self.subframe1, state='normal') 

169 self.usr_input.grid(column=0, row=0, sticky='nesw', padx=3, pady=3) 

170 self.usr_input.bind("<Return>", self.bound_enter) 

171 self.usr_input.config(width=5) 

172 

173 self.conversation_lbl = tttk.Label( 

174 self.subframe1, anchor=tkinter.E, text='Conversation') 

175 self.conversation_lbl.grid( 

176 column=0, row=1, sticky='nesw', padx=3, pady=3) 

177 self.conversation_lbl.config(width=50) 

178 

179 self.conversation = ScrolledText.ScrolledText( 

180 self.subframe1, state='disabled') 

181 self.conversation.grid(column=0, row=2, columnspan=2, 

182 sticky='nesw', padx=3, pady=3) 

183 self.conversation.config(width=50) 

184 self.bind("<<thread_fini>>", self.receive_speech) 

185 self._waiting = False 

186 

187 def bound_enter(self, *args): 

188 """ 

189 Returned was pressed. 

190 """ 

191 self.get_response() 

192 

193 def get_response(self): 

194 """ 

195 Get the text in the message box. 

196 Retrieve the answer. 

197 """ 

198 if self._waiting: 

199 self.fLOG("[TkinterMokadi] already waiting") 

200 return 

201 user_text = self.usr_input.get() 

202 self.usr_input.delete(0, tkinter.END) 

203 

204 if len(user_text) == 0: 

205 # We record the speech instead. 

206 self.start_speech_recording() 

207 else: 

208 user_input = MokadiMessage("MOKADI " + user_text, 1) 

209 self.process(user_input) 

210 

211 def start_speech_recording(self): 

212 """ 

213 Launches a thread which record the speech. 

214 """ 

215 self.conversation_lbl.config(text='Parlez (< 5s) et attendez.') 

216 self.conversation_lbl.update_idletasks() 

217 self._waiting = True 

218 th = ThreadSpeech(self, self._subkey_speech, self.fLOG) 

219 th.start() 

220 

221 def start_speech_listening(self): 

222 """ 

223 Launches a thread which record the speech. 

224 """ 

225 if hasattr(self, "thread_listen") and self.thread_listen is not None: 

226 self.thread_listen.stop_listening() 

227 self.listen.config(text="Ecouter") 

228 self.thread_listen = None 

229 else: 

230 self.listen.config(text="Arrêter l'écoute") 

231 th = ThreadListen(self, self.fLOG, self._folder) 

232 th.start() 

233 self.thread_listen = th 

234 self.bind("<<Listener>>", self.from_listener) 

235 

236 def from_listener(self, *args): 

237 """ 

238 Listener tells to start. 

239 """ 

240 speak("Je t'écoute.") 

241 self.get_response() 

242 

243 def receive_speech(self, *args): 

244 """ 

245 Reveices the recognized speech. 

246 """ 

247 self.conversation_lbl.config(text='Conversation') 

248 self.conversation_lbl.update_idletasks() 

249 

250 data = self.queue.get() 

251 self.queue.task_done() 

252 self._waiting = False 

253 

254 if isinstance(data, tuple): 

255 data = data[0] 

256 

257 self.usr_input.insert(0, data) 

258 self.usr_input.event_generate("<Return>") 

259 

260 def process(self, user_input): 

261 """ 

262 Process an input. 

263 

264 @param user_input user input 

265 """ 

266 self.conversation['state'] = 'normal' 

267 self.conversation.insert( 

268 tkinter.END, "\n\nVous : " + user_input.message) 

269 iter = self._mokadi.process(user_input, exc=False) 

270 self.fLOG("[TkinterMokadi] sent:", user_input) 

271 

272 for info in iter: 

273 speakable = [] 

274 

275 self.fLOG("[TkinterMokadi] received:", info, 

276 info.has_sound, info.has_image) 

277 

278 if info.status == "ok": 

279 if info.has_sound: 

280 play_speech(info.sound) 

281 elif info.has_image: 

282 if isinstance(info.image, str) and not info.image.startswith("http"): 

283 monimage = Image.open(info.image) 

284 photo = ImageTk.PhotoImage(monimage) 

285 label = tkinter.Label(self.subframe2, image=photo) 

286 label.image = photo 

287 label.grid(column=1, row=0) 

288 if info.info: 

289 self.conversation.insert( 

290 tkinter.END, "\nMokadi : " + info.info) 

291 if not info.has_sound: 

292 speakable.append(info.info) 

293 elif info.status == "error": 

294 self.conversation.insert( 

295 tkinter.END, "\nMokadi : " + info.info) 

296 speakable.append(info.info) 

297 

298 if self._speak and len(speakable) > 0: 

299 for text in speakable: 

300 speak(text) 

301 

302 self.conversation['state'] = 'disabled' 

303 self.conversation.see(tkinter.END) 

304 

305 

306def gui_mokadi(fLOG=None, folder_slides=None): 

307 """ 

308 Launches the application. 

309 

310 There is a bug somewhere. Taking a picture fails if the vocal synthesis 

311 runs first. We need to take a picture first. 

312 """ 

313 take_picture() 

314 user = get_password("gmail", os.environ["COMPUTERNAME"] + "user") 

315 pwd = get_password("gmail", os.environ["COMPUTERNAME"] + "pwd") 

316 server = "imap.gmail.com" 

317 subkey_news = get_password( 

318 "cogser", os.environ["COMPUTERNAME"] + "news") 

319 subkey_emo = get_password( 

320 "cogser", os.environ["COMPUTERNAME"] + "emotions") 

321 subkey_speech = get_password( 

322 "cogser", os.environ["COMPUTERNAME"] + "voicereco") 

323 

324 if folder_slides is None: 

325 folder_slides = os.path.abspath(os.path.dirname(__file__)) 

326 folder_slides = os.path.join(folder_slides, "demo") 

327 

328 folder = os.path.abspath("temp_mokadi_folder") 

329 if fLOG is not None: 

330 fLOG("[gui_mokadi] saving into", folder) 

331 if not os.path.exists(folder): 

332 os.mkdir(folder) 

333 clog = CustomLog(folder) 

334 

335 actions = [MokadiActionSlides(folder=folder_slides, fLOG=fLOG), 

336 MokadiActionMail(user=user, pwd=pwd, server=server, fLOG=fLOG), 

337 MokadiActionNews(subkey_news, fLOG=fLOG), 

338 MokadiActionEmotion(subkey_emo, folder, fLOG=fLOG), 

339 MokadiActionConversation(fLOG=fLOG), 

340 ] 

341 

342 engine = MokadiEngine(folder, clog, actions, MokadiGrammar_frParser, 

343 MokadiGrammar_frLexer, MokadiGrammar_frListener) 

344 

345 tk = tkinter.Tk() 

346 tk.iconbitmap(os.path.join(os.path.dirname(__file__), 'project_ico.ico')) 

347 tk.title("Mokadi") 

348 window = TkinterMokadi(tk, engine, vocal=True, subkey_speech=subkey_speech) 

349 tk.mainloop() 

350 if hasattr(window, "thread_listen") and window.thread_listen is not None: 

351 window.thread_listen.stop_listening() 

352 fLOG("Stop listening.") 

353 fLOG("END.")