3681

pending completion

Build # 3681

Build Type

push

travis-ci-com

Committed by

web-flow

Commit Message

Merge pull request #1105 from stanford-oval/wip/nrser/cache-hackery

Optional NLU and TTS caching with Redis

Run Details

2171 of 3973 branches covered (54.64%)

Branch coverage included in aggregate %.

116 of 205 new or added lines in 5 files covered. (56.59%)

3 existing lines in 2 files now uncovered.

8324 of 12074 relevant lines covered (68.94%)

580.11 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

55.09

/src/nlp/voice/index.ts

// -*- mode: typescript; indent-tabs-mode: nil; js-basic-offset: 4 -*-
//
// This file is part of Almond
//
// Copyright 2020 The Board of Trustees of the Leland Stanford Junior University
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Author: Euirim Choi <euirim@cs.stanford.edu>
//         Giovanni Campagna <gcampagn@cs.stanford.edu>

import { createHash } from "crypto";
import WebSocket from "ws";
import express from "express";
import multer from "multer";
import * as os from "os";
import * as iv from "../../util/input_validation";
import * as I18n from "../../util/i18n";

import { SpeechToText, TextToSpeech } from "./backend-microsoft";
import runNLU from "../nlu";
import { getRedisClient, hasRedis } from "../../util/redis";

const upload = multer({ dest: os.tmpdir() });

const router = express.Router();

const textToSpeech = new TextToSpeech();

async function streamSTT(ws : WebSocket, req : express.Request) {
    if (!I18n.get(req.params.locale, false)) {
        await ws.send(JSON.stringify({ error: "Unsupported language" }));
        await ws.close();
        return;
    }

    /* WS close codes:
     * 1000 - indicates a normal closure, meaning that the purpose for which
     * the connection was established has been fulfilled.
     *
     * 1002 - indicates that an endpoint is terminating the connection due to a
     * protocol error.
     *
     * 1003 - indicates that an endpoint is terminating the connection because
     * it has received a type of data it cannot accept.
     */
    function errorClose(e : { error : string }, code = 1002) {
        if (ws.readyState === 1) {
            // OPEN
            ws.send(JSON.stringify(e));
            ws.close(code);
        }
    }

    const sessionTimeout = setTimeout(() => {
        errorClose({ error: "Session timeout" });
    }, 5000);

    ws.on("close", () => {
        if (sessionTimeout) clearTimeout(sessionTimeout);
    });

    function initialPacket(msg : string) {
        ws.removeListener("message", initialPacket);
        clearTimeout(sessionTimeout);

        let parsed : { ver : number };
        try {
            parsed = JSON.parse(msg);
        } catch(e) {
            errorClose({ error: "Malformed initial packet: " + e.message });
            return;
        }

        if (parsed.ver && parsed.ver === 1) {
            const stt = new SpeechToText(req.params.locale);
            stt.recognizeStream(ws)
                .then((text) => {
                    const result = { result: "ok", text: text };
                    ws.send(JSON.stringify(result));
                    ws.close(1000);
                })
                .catch((e) => {
                    errorClose(e, e.status !== 400 ? 1003 : 1000);
                });
        } else {
            errorClose({ error: "Unsupported protocol" });
        }
    }

    ws.on("message", initialPacket);
}

function restSTT(
    req : express.Request,
    res : express.Response,
    next : express.NextFunction
) {
    if (!req.file) {
        iv.failKey(req, res, "audio", { json: true });
        return;
    }
    if (!I18n.get(req.params.locale, false)) {
        res.status(404).json({ error: "Unsupported language" });
        return;
    }

    const stt = new SpeechToText(req.params.locale);
    stt.recognizeOnce(req.file.path)
        .then((text) => {
            res.json({
                result: "ok",
                text: text,
            });
        })
        .catch(next);
}

const NLU_METADATA_KEYS = {
    store: "?string",
    access_token: "?string",
    thingtalk_version: "?string",
    limit: "?integer",
    expect: "?string",
    choices: "?array",
    context: "?string",
    entities: "?object",
    tokenized: "boolean",
    skip_typechecking: "boolean",
    developer_key: "?string",
} as const;

async function restSTTAndNLU(
    req : express.Request,
    res : express.Response,
    next : express.NextFunction
) {
    if (!req.file) {
        iv.failKey(req, res, "audio", { json: true });
        return;
    }
    let metadata;
    try {
        metadata = JSON.parse(req.body.metadata);
    } catch(e) {
        iv.failKey(req, res, "metadata", { json: true });
        return;
    }
    for (const key in NLU_METADATA_KEYS) {
        if (
            !iv.checkKey(
                metadata[key],
                NLU_METADATA_KEYS[key as keyof typeof NLU_METADATA_KEYS]
            )
        ) {
            iv.failKey(req, res, key, { json: true });
            return;
        }
    }

    if (!I18n.get(req.params.locale, false)) {
        res.status(404).json({ error: "Unsupported language" });
        return;
    }

    const stt = new SpeechToText(req.params.locale);
    const text = await stt.recognizeOnce(req.file.path);

    const result = await runNLU(
        text,
        req.params,
        req.body,
        res.app.service,
        res
    );
    if (result === undefined) return;

    result.text = text;
    res.json(result);
}

async function ttsCached(req : express.Request, res : express.Response) {
    if (!I18n.get(req.params.locale, false)) {
        res.status(404).json({ error: "Unsupported language" });
        return;
    }
    const locale = req.params.locale;
    const gender = req.params.gender === "female" ? "female" : "male";
    const text = req.body.text;

    const argsSig = createHash("md5")
        .update(JSON.stringify([locale, gender, text]))
        .digest("hex");
    const cacheKey = `TextToSpeech.request:${argsSig}`;
    const redisClient = await getRedisClient();
    const cacheValue = await redisClient.GET_BUFFER(cacheKey);

    if (cacheValue === null) {
        console.log(`CACHE MISS ${cacheKey}`);
        const stream = await textToSpeech.request(locale, gender, text);
        if (stream.statusCode === 200) res.set("Content-Type", "audio/x-wav");
        stream.pipe(res);
        const chunks : Buffer[] = [];
        stream.on("data", (chunk) => {
            if (Buffer.isBuffer(chunk)) chunks.push(chunk);
            else console.warn(`Received non-Buffer ${typeof chunk}`);
        });
        stream.on("end", () => {
            if (chunks.length === 0) {
                console.warn(`No chunks added!`);
                return;
            }
            const buffer = Buffer.concat(chunks);
            console.log(`CACHE SET ${cacheKey}`);
            redisClient.SET(cacheKey, buffer);
        });
    } else {
        console.log(`CACHE HIT ${cacheKey}`);
        res.set("Content-Type", "audio/x-wav");
        res.end(cacheValue);
    }
}

async function ttsNoCache(req : express.Request, res : express.Response) {
    if (!I18n.get(req.params.locale, false)) {
        res.status(404).json({ error: "Unsupported language" });
        return;
    }
    const locale = req.params.locale;
    const gender = req.params.gender === "female" ? "female" : "male";
    const text = req.body.text;

    const stream = await textToSpeech.request(locale, gender, text);
    if (stream.statusCode === 200) res.set("Content-Type", "audio/x-wav");
    stream.pipe(res);
}

function ttspost(
    req : express.Request,
    res : express.Response,
    next : express.NextFunction
) {
    if (hasRedis()) ttsCached(req, res).catch(next);
    else ttsNoCache(req, res).catch(next);
}

function ttsget(
    req : express.Request,
    res : express.Response,
    next : express.NextFunction
) {
    if (hasRedis()) ttsCached(req, res).catch(next);
    else ttsNoCache(req, res).catch(next);
}

router.ws("/:locale/voice/stream", streamSTT);

router.post("/:locale/voice/stt", upload.single("audio"), restSTT);
router.post(
    "/:locale/voice/query",
    upload.single("audio"),
    iv.validatePOST({ metadata: "string" }),
    restSTTAndNLU
);
router.post(
    "/:locale/voice/tts",
    iv.validatePOST(
        {
            gender: /^(|male|female)$/,
            text: "string",
        },
        { json: true }
    ),
    ttspost
);
router.get(
    "/:locale/voice/tts",
    iv.validateGET(
        {
            gender: /^(|male|female)$/,
            text: "string",
        },
        { json: true }
    ),
    ttsget
);

// provide identical API keyed off to :model_tag, so people can change the
// NL_SERVER_URL to include the model tag

router.post("/@:model_tag/:locale/voice/stt", upload.single("audio"), restSTT);
router.post(
    "/@:model_tag/:locale/voice/query",
    upload.single("audio"),
    iv.validatePOST({ metadata: "string" }, { json: true }),
    restSTTAndNLU
);
router.post(
    "/@:model_tag/:locale/voice/tts",
    iv.validatePOST(
        {
            gender: /^(|male|female)$/,
            text: "string",
        },
        { json: true }
    ),
    ttspost
);
router.get(
    "/@:model_tag/:locale/voice/tts",
    iv.validateGET(
        {
            gender: /^(|male|female)$/,
            text: "string",
        },
        { json: true }
    ),
    ttsget
);

export default router;

1	// -- mode: typescript; indent-tabs-mode: nil; js-basic-offset: 4 --
2	//
3	// This file is part of Almond
4	//
5	// Copyright 2020 The Board of Trustees of the Leland Stanford Junior University
6	//
7	// Licensed under the Apache License, Version 2.0 (the "License");
8	// you may not use this file except in compliance with the License.
9	// You may obtain a copy of the License at
10	//
11	// http://www.apache.org/licenses/LICENSE-2.0
12	//
13	// Unless required by applicable law or agreed to in writing, software
14	// distributed under the License is distributed on an "AS IS" BASIS,
15	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16	// See the License for the specific language governing permissions and
17	// limitations under the License.
18	//
19	// Author: Euirim Choi <euirim@cs.stanford.edu>
20	// Giovanni Campagna <gcampagn@cs.stanford.edu>
21
22	import { createHash } from "crypto";	1✔
23	import WebSocket from "ws";
24	import express from "express";	1✔
25	import multer from "multer";	1✔
26	import * as os from "os";	1✔
27	import * as iv from "../../util/input_validation";	1✔
28	import * as I18n from "../../util/i18n";	1✔
29
30	import { SpeechToText, TextToSpeech } from "./backend-microsoft";	1✔
31	import runNLU from "../nlu";	1✔
32	import { getRedisClient, hasRedis } from "../../util/redis";	1✔
33
34	const upload = multer({ dest: os.tmpdir() });	1✔
35
36	const router = express.Router();	1✔
37
38	const textToSpeech = new TextToSpeech();	1✔
39
40	async function streamSTT(ws : WebSocket, req : express.Request) {
41	if (!I18n.get(req.params.locale, false)) {	4!
NEW 42	await ws.send(JSON.stringify({ error: "Unsupported language" }));	×
43	await ws.close();	×
44	return;	×
45	}
46
47	/* WS close codes:
48	* 1000 - indicates a normal closure, meaning that the purpose for which
49	* the connection was established has been fulfilled.
50	*
51	* 1002 - indicates that an endpoint is terminating the connection due to a
52	* protocol error.
53	*
54	* 1003 - indicates that an endpoint is terminating the connection because
55	* it has received a type of data it cannot accept.
56	*/
57	function errorClose(e : { error : string }, code = 1002) {	2✔
58	if (ws.readyState === 1) {	3!
59	// OPEN
60	ws.send(JSON.stringify(e));	3✔
61	ws.close(code);	3✔
62	}
63	}
64
65	const sessionTimeout = setTimeout(() => {	4✔
66	errorClose({ error: "Session timeout" });	1✔
67	}, 5000);
68
69	ws.on("close", () => {	4✔
70	if (sessionTimeout) clearTimeout(sessionTimeout);	4!
71	});
72
73	function initialPacket(msg : string) {
74	ws.removeListener("message", initialPacket);	3✔
75	clearTimeout(sessionTimeout);	3✔
76
77	let parsed : { ver : number };
78	try {	3✔
79	parsed = JSON.parse(msg);	3✔
80	} catch(e) {
NEW 81	errorClose({ error: "Malformed initial packet: " + e.message });	×
NEW 82	return;	×
83	}
84
85	if (parsed.ver && parsed.ver === 1) {	3✔
86	const stt = new SpeechToText(req.params.locale);	2✔
87	stt.recognizeStream(ws)	2✔
88	.then((text) => {
89	const result = { result: "ok", text: text };	1✔
90	ws.send(JSON.stringify(result));	1✔
91	ws.close(1000);	1✔
92	})
93	.catch((e) => {
94	errorClose(e, e.status !== 400 ? 1003 : 1000);	1!
95	});
96	} else {
97	errorClose({ error: "Unsupported protocol" });	1✔
98	}
99	}
100
101	ws.on("message", initialPacket);	4✔
102	}
103
104	function restSTT(
105	req : express.Request,
106	res : express.Response,
107	next : express.NextFunction
108	) {
109	if (!req.file) {	2✔
110	iv.failKey(req, res, "audio", { json: true });	1✔
111	return;	1✔
112	}
113	if (!I18n.get(req.params.locale, false)) {	1!
NEW 114	res.status(404).json({ error: "Unsupported language" });	×
115	return;	×
116	}
117
118	const stt = new SpeechToText(req.params.locale);	1✔
119	stt.recognizeOnce(req.file.path)	1✔
120	.then((text) => {
121	res.json({	1✔
122	result: "ok",
123	text: text,
124	});
125	})
126	.catch(next);
127	}
128
129	const NLU_METADATA_KEYS = {	1✔
130	store: "?string",
131	access_token: "?string",
132	thingtalk_version: "?string",
133	limit: "?integer",
134	expect: "?string",
135	choices: "?array",
136	context: "?string",
137	entities: "?object",
138	tokenized: "boolean",
139	skip_typechecking: "boolean",
140	developer_key: "?string",
141	} as const;
142
143	async function restSTTAndNLU(
144	req : express.Request,
145	res : express.Response,
146	next : express.NextFunction
147	) {
148	if (!req.file) {	1!
NEW 149	iv.failKey(req, res, "audio", { json: true });	×
150	return;	×
151	}
152	let metadata;
153	try {	1✔
154	metadata = JSON.parse(req.body.metadata);	1✔
155	} catch(e) {
NEW 156	iv.failKey(req, res, "metadata", { json: true });	×
157	return;	×
158	}
159	for (const key in NLU_METADATA_KEYS) {	1✔
160	if (	11!
161	!iv.checkKey(
162	metadata[key],
163	NLU_METADATA_KEYS[key as keyof typeof NLU_METADATA_KEYS]
164	)
165	) {
166	iv.failKey(req, res, key, { json: true });	×
167	return;	×
168	}
169	}
170
171	if (!I18n.get(req.params.locale, false)) {	1!
NEW 172	res.status(404).json({ error: "Unsupported language" });	×
173	return;	×
174	}
175
176	const stt = new SpeechToText(req.params.locale);	1✔
177	const text = await stt.recognizeOnce(req.file.path);	1✔
178
179	const result = await runNLU(	1✔
180	text,
181	req.params,
182	req.body,
183	res.app.service,
184	res
185	);
186	if (result === undefined) return;	1!
187
188	result.text = text;	1✔
189	res.json(result);	1✔
190	}
191
192	async function ttsCached(req : express.Request, res : express.Response) {
UNCOV 193	if (!I18n.get(req.params.locale, false)) {	×
NEW 194	res.status(404).json({ error: "Unsupported language" });	×
195	return;	×
196	}
NEW 197	const locale = req.params.locale;	×
NEW 198	const gender = req.params.gender === "female" ? "female" : "male";	×
NEW 199	const text = req.body.text;	×
200
NEW 201	const argsSig = createHash("md5")	×
202	.update(JSON.stringify([locale, gender, text]))
203	.digest("hex");
NEW 204	const cacheKey = `TextToSpeech.request:${argsSig}`;	×
NEW 205	const redisClient = await getRedisClient();	×
NEW 206	const cacheValue = await redisClient.GET_BUFFER(cacheKey);	×
207
NEW 208	if (cacheValue === null) {	×
NEW 209	console.log(`CACHE MISS ${cacheKey}`);	×
NEW 210	const stream = await textToSpeech.request(locale, gender, text);	×
NEW 211	if (stream.statusCode === 200) res.set("Content-Type", "audio/x-wav");	×
UNCOV 212	stream.pipe(res);	×
NEW 213	const chunks : Buffer[] = [];	×
NEW 214	stream.on("data", (chunk) => {	×
NEW 215	if (Buffer.isBuffer(chunk)) chunks.push(chunk);	×
NEW 216	else console.warn(`Received non-Buffer ${typeof chunk}`);	×
217	});
NEW 218	stream.on("end", () => {	×
NEW 219	if (chunks.length === 0) {	×
NEW 220	console.warn(`No chunks added!`);	×
NEW 221	return;	×
222	}
NEW 223	const buffer = Buffer.concat(chunks);	×
NEW 224	console.log(`CACHE SET ${cacheKey}`);	×
NEW 225	redisClient.SET(cacheKey, buffer);	×
226	});
227	} else {
NEW 228	console.log(`CACHE HIT ${cacheKey}`);	×
NEW 229	res.set("Content-Type", "audio/x-wav");	×
NEW 230	res.end(cacheValue);	×
231	}
232	}
233
234	async function ttsNoCache(req : express.Request, res : express.Response) {
235	if (!I18n.get(req.params.locale, false)) {	1!
NEW 236	res.status(404).json({ error: "Unsupported language" });	×
237	return;	×
238	}
239	const locale = req.params.locale;	1✔
240	const gender = req.params.gender === "female" ? "female" : "male";	1!
241	const text = req.body.text;	1✔
242
243	const stream = await textToSpeech.request(locale, gender, text);	1✔
244	if (stream.statusCode === 200) res.set("Content-Type", "audio/x-wav");	1!
245	stream.pipe(res);	1✔
246	}
247
248	function ttspost(
249	req : express.Request,
250	res : express.Response,
251	next : express.NextFunction
252	) {
253	if (hasRedis()) ttsCached(req, res).catch(next);	1!
254	else ttsNoCache(req, res).catch(next);	1✔
255	}
256
257	function ttsget(
258	req : express.Request,
259	res : express.Response,
260	next : express.NextFunction
261	) {
NEW 262	if (hasRedis()) ttsCached(req, res).catch(next);	×
NEW 263	else ttsNoCache(req, res).catch(next);	×
264	}
265
266	router.ws("/:locale/voice/stream", streamSTT);	1✔
267
268	router.post("/:locale/voice/stt", upload.single("audio"), restSTT);	1✔
269	router.post(	1✔
270	"/:locale/voice/query",
271	upload.single("audio"),
272	iv.validatePOST({ metadata: "string" }),
273	restSTTAndNLU
274	);
275	router.post(	1✔
276	"/:locale/voice/tts",
277	iv.validatePOST(
278	{
279	gender: /^(\|male\|female)$/,
280	text: "string",
281	},
282	{ json: true }
283	),
284	ttspost
285	);
286	router.get(	1✔
287	"/:locale/voice/tts",
288	iv.validateGET(
289	{
290	gender: /^(\|male\|female)$/,
291	text: "string",
292	},
293	{ json: true }
294	),
295	ttsget
296	);
297
298	// provide identical API keyed off to :model_tag, so people can change the
299	// NL_SERVER_URL to include the model tag
300
301	router.post("/@:model_tag/:locale/voice/stt", upload.single("audio"), restSTT);	1✔
302	router.post(	1✔
303	"/@:model_tag/:locale/voice/query",
304	upload.single("audio"),
305	iv.validatePOST({ metadata: "string" }, { json: true }),
306	restSTTAndNLU
307	);
308	router.post(	1✔
309	"/@:model_tag/:locale/voice/tts",
310	iv.validatePOST(
311	{
312	gender: /^(\|male\|female)$/,
313	text: "string",
314	},
315	{ json: true }
316	),
317	ttspost
318	);
319	router.get(	1✔
320	"/@:model_tag/:locale/voice/tts",
321	iv.validateGET(
322	{
323	gender: /^(\|male\|female)$/,
324	text: "string",
325	},
326	{ json: true }
327	),
328	ttsget
329	);
330
331	export default router;	1✔

stanford-oval / almond-cloud / 3681

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous