Coverage for askbob\server\voice.py : 100%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from askbob.speech.listener.file import FileUtteranceService
2from askbob.speech.transcriber import Transcriber, TranscriptionEvent
3from sanic import Sanic
4from sanic.response import json
5import tempfile
6import os
7import logging
10def voice_routes(app: Sanic, responder, config: dict):
11 # Make transcriber
12 if 'Listener' not in config:
13 raise RuntimeError(
14 "Missing Listener section in the runtime configuration file.")
16 if 'model' not in config['Listener']:
17 raise RuntimeError(
18 "Missing Listener.model in the runtime configuration file.")
20 model_path = config['Listener']['model']
21 scorer_path = config['Listener'].get('scorer', '')
23 transcriber = Transcriber(
24 model=model_path, scorer=scorer_path, us=None)
26 # Setup voice query handler
27 @app.route("/voicequery", methods=['POST', 'OPTIONS'])
28 async def voice(request):
30 sender = request.form.get('sender')
31 if not sender:
32 return json({
33 "error": "A 'sender' must be provided."
34 })
36 if not sender.isprintable():
37 return json({
38 "error": "The sender must contain printable characters."
39 })
41 if 'speech' not in request.files:
42 return json({
43 "error": "No speech file provided."
44 })
46 if len(request.files['speech']) != 1:
47 return json({
48 "error": "Too many speech files uploaded."
49 })
51 file = request.files['speech'][0]
52 if file.name.split('.')[-1] != 'wav' or file.type != 'audio/wav':
53 return json({
54 "error": "Incorrect speech file type - it must be audio/wav."
55 })
57 if len(file.body) >= 10485760:
58 return json({
59 "error": "The speech file must be <10MiB."
60 })
62 f = tempfile.NamedTemporaryFile(delete=False)
63 f.write(file.body)
64 f.close()
66 try:
67 transcriber.us = FileUtteranceService(filename=f.name, aggressiveness=config['Listener'].getint(
68 'aggressiveness', fallback=1))
69 except Exception as e:
70 transcriber.us = None
71 return json({
72 "error": str(e)
73 })
75 for state, text in transcriber.transcribe():
76 if state == TranscriptionEvent.END_UTTERANCE:
77 os.unlink(f.name)
78 transcriber.us = None
79 if not text or text.isspace():
80 return json({
81 "query": "",
82 "messages": []
83 })
85 return json({
86 "query": text,
87 "messages": [
88 response async for response in responder.handle(text, sender)]
89 })
91 transcriber.us = None
92 return json({
93 "error": "Speech transcription failed."
94 })