2013

pending completion

Build # 2013

Build Type

Pull #619

travis-ci-com

Committed by

web-flow

Commit Message

Merge 0530df35c into 0b153da3b

Pull Request Pull Request #619: Implement stream based STT through almond-cloud WS

Run Details

12387 of 26066 branches covered (47.52%)

Branch coverage included in aggregate %.

6 of 143 new or added lines in 3 files covered. (4.2%)

19 existing lines in 6 files now uncovered.

28860 of 48969 relevant lines covered (58.94%)

451844.5 hits per line

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

3.92

/lib/speech/speech_recognizer.js

// -*- mode: js; indent-tabs-mode: nil; js-basic-offset: 4 -*-
//
// This file is part of Genie
//
// 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: Giovanni Campagna <gcampagn@cs.stanford.edu>


import * as events from 'events';
import * as Stream from 'stream';

import * as uuid from 'uuid';
import WebSocket from 'ws';

const URL = 'https://almond-nl.stanford.edu';

class SpeechRequest extends Stream.Writable {
    constructor(stream, initialBuffer, vad) {
        // this is important, as the protocol does not allow chunks larger than 8192 bytes
        super({ highWaterMark: 8192 });

        this._stream = stream;

        this._requestId = uuid.v4().replace(/-/g, '');
        this._endDetected = false;
        this._ended = false;
        this._started = false;

        //this._debugFile = fs.createWriteStream('out_' + process.pid + '_' + (i++) + '.wav');
        this._vad = vad;
        this.consecutiveSilence = 0;

        // all chunks received before the connection is ready are buffered here
        this._bufferedMessages = [];

        this.write(initialBuffer);
        this._stream.pipe(this);
        this._piped = true;

        this.on('error', () => {
            if (!this._piped)
                return;
            this._piped = false;
            this._stream.unpipe(this);
        });
    }

    start(connection) {
        this._started = true;
        this._connection = connection;
        this._endTimeout = setTimeout(() => this.end(), 150000);

        this._listener = this._handleMessage.bind(this);
        this._connection.on('message', this._listener);

        for (let message of this._bufferedMessages)
            this._connection.send(message, { binary: true });
        this._bufferedMessages = [];
    }

    _finish(callback) {
        // readable stream ended
        this._piped = false;
        this.end();
        callback();
    }

    end() {
        if (this._piped) {
            this._piped = false;
            this._stream.unpipe(this);
        }

        if (this._ended)
            return;

        // end() before start() indicates an error connecting to the server (e.g.
        // access token error)
        if (!this._started) {
            this._ended = true;
            return;
        }

        clearTimeout(this._endTimeout);

        this._ended = true;
        //this._debugFile.end();
    }

    _handleMessage(msg) {
        //console.log('Received message');
        //console.log(msg);
        try {
            msg = JSON.parse(msg);
        } catch(e) {
            this.emit('error', e);
            this.end();
            return;
        }

        if (msg.status === 400 && msg.code === "E_NO_MATCH")
            this.emit('done', "NoMatch");
        else if (msg.result === 'ok')
            this.emit('done', "Success", msg.text);
        else
            this.emit('error', msg);
        this.end();
    }

    _write(chunk, encoding, callback) {
        if (this._ended || this._endDetected) {
            callback();
            return;
        }
        //console.log('Sending chunk of length ' + chunk.length);
        if (this._connection && this._connection.readyState === 1) {
            // OPEN
            if (this._vad && chunk.length === 320) {
                if (this._vad.process(chunk))
                    this.consecutiveSilence = 0;
                else
                    this.consecutiveSilence++;
            }
            if (this.consecutiveSilence === 96) {
                this._endDetected = true;
                console.log("VAD threshold reached", this.consecutiveSilence);
                this._connection.send(undefined, {}, (err) => callback(err));
            } else {
                this._connection.send(chunk, { binary: true }, (err) => callback(err));
            }
        } else {
            this._bufferedMessages.push(chunk);
            callback();
        }
    }
}

export default class SpeechRecognizer extends events.EventEmitter {
    constructor(options = {}) {
        super();
        this._language = options.locale || 'en-US';
        this._baseUrl = options.nlUrl || URL;
        this._connection = null;
        this._vad = options.vad || null;
    }

    close() {
        if (!this._connection)
            return;
        this._connection.close();
        this._connection = null;
    }

    _doConnect() {
        let url = this._baseUrl + '/' + this._language + '/voice/stream';
        let connection = new WebSocket(url, {
            perMessageDeflate: true
        });
        return new Promise((callback, errback) => {
            connection.on('unexpected-response', (req, res) => {
                errback(new Error(res.statusMessage));
            });
            connection.on('open', () => {
                this._connection = connection;
                console.log('STT connection opened');
                this._connection.send(JSON.stringify({ ver: 1 }));
                callback(connection);
            });
            connection.on('close', (code, reason) => {
                if (code !== 1000) // 1000 = normal closure (eg timeout, or we closed on our side)
                    console.log('Connection to STT service closed: ' + code + ' ' + reason);
                this._connection = null;
            });
            connection.on('error', (e) => {
                this._connection = null;
                if (e.code === 'ECONNRESET')
                    console.log('Error on STT service: Connection Reset');
                else
                    this.emit('error', e);
            });
        });
    }

    _ensureConnection() {
        if (this._connection)
            return Promise.resolve(this._connection);
        else
            return this._doConnect();
    }

    request(stream, initialBuffer) {
        let req = new SpeechRequest(stream, initialBuffer, this._vad);
        this._ensureConnection().then((connection) => {
            req.start(connection);
        }).catch((e) => {
            req.end();
            this.emit('error', e);
        });
        req.on('error', (e) => this.emit('error', e));
        return req;
    }
}

1	// -- mode: js; indent-tabs-mode: nil; js-basic-offset: 4 --
2	//
3	// This file is part of Genie
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: Giovanni Campagna <gcampagn@cs.stanford.edu>
20
21
22	import * as events from 'events';	2✔
23	import * as Stream from 'stream';	2✔
24
25	import * as uuid from 'uuid';	2✔
26	import WebSocket from 'ws';	2✔
27
28	const URL = 'https://almond-nl.stanford.edu';	2✔
29
30	class SpeechRequest extends Stream.Writable {
31	constructor(stream, initialBuffer, vad) {
32	// this is important, as the protocol does not allow chunks larger than 8192 bytes
33	super({ highWaterMark: 8192 });	×
34
35	this._stream = stream;	×
36
37	this._requestId = uuid.v4().replace(/-/g, '');	×
38	this._endDetected = false;	×
39	this._ended = false;	×
40	this._started = false;	×
41
42	//this._debugFile = fs.createWriteStream('out_' + process.pid + '_' + (i++) + '.wav');
NEW 43	this._vad = vad;	×
NEW 44	this.consecutiveSilence = 0;	×
45
46	// all chunks received before the connection is ready are buffered here
47	this._bufferedMessages = [];	×
48
49	this.write(initialBuffer);	×
50	this._stream.pipe(this);	×
51	this._piped = true;	×
52
53	this.on('error', () => {	×
54	if (!this._piped)	×
55	return;	×
56	this._piped = false;	×
57	this._stream.unpipe(this);	×
58	});
59	}
60
61	start(connection) {
62	this._started = true;	×
63	this._connection = connection;	×
64	this._endTimeout = setTimeout(() => this.end(), 150000);	×
65
66	this._listener = this._handleMessage.bind(this);	×
67	this._connection.on('message', this._listener);	×
68
69	for (let message of this._bufferedMessages)	×
70	this._connection.send(message, { binary: true });	×
71	this._bufferedMessages = [];	×
72	}
73
74	_finish(callback) {
75	// readable stream ended
76	this._piped = false;	×
77	this.end();	×
78	callback();	×
79	}
80
81	end() {
82	if (this._piped) {	×
83	this._piped = false;	×
84	this._stream.unpipe(this);	×
85	}
86
87	if (this._ended)	×
88	return;	×
89
90	// end() before start() indicates an error connecting to the server (e.g.
91	// access token error)
92	if (!this._started) {	×
93	this._ended = true;	×
94	return;	×
95	}
96
97	clearTimeout(this._endTimeout);	×
98
UNCOV 99	this._ended = true;	×
100	//this._debugFile.end();
101	}
102
103	_handleMessage(msg) {
104	//console.log('Received message');
105	//console.log(msg);
NEW 106	try {	×
NEW 107	msg = JSON.parse(msg);	×
108	} catch(e) {
NEW 109	this.emit('error', e);	×
NEW 110	this.end();	×
UNCOV 111	return;	×
112	}
113
NEW 114	if (msg.status === 400 && msg.code === "E_NO_MATCH")	×
NEW 115	this.emit('done', "NoMatch");	×
NEW 116	else if (msg.result === 'ok')	×
NEW 117	this.emit('done', "Success", msg.text);	×
118	else
NEW 119	this.emit('error', msg);	×
NEW 120	this.end();	×
121	}
122
123	_write(chunk, encoding, callback) {
124	if (this._ended \|\| this._endDetected) {	×
125	callback();	×
126	return;	×
127	}
128	//console.log('Sending chunk of length ' + chunk.length);
UNCOV 129	if (this._connection && this._connection.readyState === 1) {	×
130	// OPEN
NEW 131	if (this._vad && chunk.length === 320) {	×
NEW 132	if (this._vad.process(chunk))	×
NEW 133	this.consecutiveSilence = 0;	×
134	else
NEW 135	this.consecutiveSilence++;	×
136	}
NEW 137	if (this.consecutiveSilence === 96) {	×
NEW 138	this._endDetected = true;	×
NEW 139	console.log("VAD threshold reached", this.consecutiveSilence);	×
NEW 140	this._connection.send(undefined, {}, (err) => callback(err));	×
141	} else {
NEW 142	this._connection.send(chunk, { binary: true }, (err) => callback(err));	×
143	}
144	} else {
NEW 145	this._bufferedMessages.push(chunk);	×
146	callback();	×
147	}
148	}
149	}
150
151	export default class SpeechRecognizer extends events.EventEmitter {	2✔
152	constructor(options = {}) {	×
153	super();	×
154	this._language = options.locale \|\| 'en-US';	×
NEW 155	this._baseUrl = options.nlUrl \|\| URL;	×
156	this._connection = null;	×
NEW 157	this._vad = options.vad \|\| null;	×
158	}
159
160	close() {
161	if (!this._connection)	×
162	return;	×
163	this._connection.close();	×
164	this._connection = null;	×
165	}
166
167	_doConnect() {
NEW 168	let url = this._baseUrl + '/' + this._language + '/voice/stream';	×
UNCOV 169	let connection = new WebSocket(url, {	×
170	perMessageDeflate: true
171	});
172	return new Promise((callback, errback) => {	×
173	connection.on('unexpected-response', (req, res) => {	×
174	errback(new Error(res.statusMessage));	×
175	});
176	connection.on('open', () => {	×
UNCOV 177	this._connection = connection;	×
NEW 178	console.log('STT connection opened');	×
NEW 179	this._connection.send(JSON.stringify({ ver: 1 }));	×
UNCOV 180	callback(connection);	×
181	});
182	connection.on('close', (code, reason) => {	×
183	if (code !== 1000) // 1000 = normal closure (eg timeout, or we closed on our side)	×
NEW 184	console.log('Connection to STT service closed: ' + code + ' ' + reason);	×
185	this._connection = null;	×
186	});
187	connection.on('error', (e) => {	×
188	this._connection = null;	×
189	if (e.code === 'ECONNRESET')	×
NEW 190	console.log('Error on STT service: Connection Reset');	×
191	else
192	this.emit('error', e);	×
193	});
194	});
195	}
196
197	_ensureConnection() {
198	if (this._connection)	×
199	return Promise.resolve(this._connection);	×
200	else
NEW 201	return this._doConnect();	×
202	}
203
204	request(stream, initialBuffer) {
NEW 205	let req = new SpeechRequest(stream, initialBuffer, this._vad);	×
206	this._ensureConnection().then((connection) => {	×
NEW 207	req.start(connection);	×
208	}).catch((e) => {
209	req.end();	×
210	this.emit('error', e);	×
211	});
212	req.on('error', (e) => this.emit('error', e));	×
213	return req;	×
214	}
215	}

stanford-oval / genie-toolkit / 2013

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