2260

pending completion

Build # 2260

Build Type

Pull #751

travis-ci-com

Committed by

web-flow

Commit Message

Merge dbc940d7e into 84110df3d

Pull Request Pull Request #751: Integrate db to save and restore conversation state

Run Details

10012 of 20271 branches covered (49.39%)

Branch coverage included in aggregate %.

8 of 16 new or added lines in 3 files covered. (50.0%)

1 existing line in 1 file now uncovered.

23915 of 40146 relevant lines covered (59.57%)

358656.13 hits per line

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

47.89

/lib/dialogue-agent/assistant_dispatcher.ts

// -*- mode: typescript; 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 assert from 'assert';
import * as Tp from 'thingpedia';
import * as events from 'events';

import Conversation, {
    ConversationDelegate,
    ConversationOptions,
    ConversationState
} from './conversation';
import { PlatformData, Message } from './protocol';
import NotificationFormatter from './notifications/formatter';
import {
    StaticNotificationBackends,
    ThingpediaNotificationBackend,
    NotificationConfig,
    NotificationDelegate,
} from './notifications';

import AppExecutor from '../engine/apps/app_executor';
import type Engine from '../engine';
import DeviceView from '../engine/devices/device_view';
import DeviceInterfaceMapper from '../engine/devices/device_interface_mapper';
import { ConversationStateRow, LocalTable } from "../engine/db";

/**
 * A conversation delegate that buffers all commands until the dialogue turn
 * is complete.
 */
class StatelessConversationDelegate implements ConversationDelegate {
    private _buffer : Message[];
    private _askSpecial : string|null;
    private _id : string;

    constructor(conversationId : string) {
        this._buffer = [];
        this._askSpecial = null;
        this._id = conversationId;
    }

    flush() {
        const buffer = this._buffer;
        const askSpecial = this._askSpecial;
        this._buffer = [];
        this._askSpecial = null;
        return {
            conversationId : this._id,
            messages: buffer,
            askSpecial: askSpecial,
        };
    }

    setHypothesis() {
        // ignore
    }

    setExpected(what : string|null) {
        assert(this._askSpecial === null);
        this._askSpecial = what;
    }

    async addMessage(message : Message) {
        this._buffer.push(message);
    }
}

export interface CommandInput {
    type : 'command';
    text : string;
}

export interface ParsedInput {
    type : 'parsed';
    json : any;
    title ?: string;
}

export interface ThingTalkInput {
    type : 'tt';
    code : string;
}

export type ConverseInput = (CommandInput | ParsedInput | ThingTalkInput) & PlatformData;

/**
 * The main controller class for interaction with the user.
 *
 * This class manages multiple conversations (independent sessions of the
 * agent with their own state), it dispatches notifications, and it
 * handles connections from API clients.
 */
export default class AssistantDispatcher extends events.EventEmitter {
    private _engine : Engine;
    private _notificationFormatter : NotificationFormatter;
    private _nluModelUrl : string|undefined;

    private _notificationOutputs : Set<NotificationDelegate>;
    private _staticNotificationBackends : Record<string, Tp.Capabilities.NotificationBackend>;
    private _dynamicNotificationBackends : DeviceInterfaceMapper<Tp.Capabilities.NotificationBackend>;
    private _conversations : Map<string, Conversation>;
    private _lastConversation : Conversation|null;
    private _conversationStateDB : LocalTable<ConversationStateRow>;

    constructor(engine : Engine, nluModelUrl : string|undefined, notificationConfig : NotificationConfig) {
        super();

        this._engine = engine;
        this._notificationFormatter = new NotificationFormatter(engine);
        this._nluModelUrl = nluModelUrl;
        this._notificationOutputs = new Set;
        this._conversations = new Map;
        this._lastConversation = null;
        this._conversationStateDB = this._engine.db.getLocalTable('conversation_state');

        this._dynamicNotificationBackends = new DeviceInterfaceMapper(
            new DeviceView(engine.devices, 'org.thingpedia.notification-provider', {}),
            (device) => new ThingpediaNotificationBackend(device));

        // initialize static notification backends
        this._staticNotificationBackends = {};
        for (const key in notificationConfig) {
            const key2 = key as keyof typeof StaticNotificationBackends;
            this._staticNotificationBackends[key] = new (StaticNotificationBackends[key2])(engine, notificationConfig[key2] as any);
        }
    }

    async start() {
        await this._notificationFormatter.initialize();
        this._dynamicNotificationBackends.start();
    }
    async stop() {
        this._dynamicNotificationBackends.stop();
    }

    /**
     * Dispatch one single command to an existing conversation.
     *
     * This is an alternative to getting the conversation and adding a
     * ConversationDelegate to it.
     * It exists for the convenience of REST API clients which do not keep
     * an open web socket.
     */
    async converse(command : ConverseInput, conversationId : string) : Promise<{
        conversationId : string;
        messages : Message[],
        askSpecial : string|null;
    }> {
        const state = await this._conversationStateDB.getOne(conversationId).then((row) => {
            if (row) {
                return {
                    history : row.history ? JSON.parse(row.history) : [],
                    dialogueState : row.dialogueState ? JSON.parse(row.dialogueState) : null,
                    lastMessageId : row.lastMessageId ? row.lastMessageId : 0
                };
            }
            return undefined;
        });

        const conversation = await this.getOrOpenConversation(conversationId, {
            showWelcome: false,
            anonymous: false,
            debug: true
        }, state);
        const delegate = new StatelessConversationDelegate(conversationId);
        await conversation.addOutput(delegate, false);

        switch (command.type) {
        case 'command':
            await conversation.handleCommand(command.text, command);
            break;
        case 'parsed':
            await conversation.handleParsedCommand(command.json, command.title, command);
            break;
        case 'tt':
            await conversation.handleThingTalk(command.code, command);
            break;
        }

        await conversation.removeOutput(delegate);
        const result = delegate.flush();
        return result;
    }

    /**
     * Add an API client that wishes to listen for notifications.
     *
     * The delegate methods will be invoked for every notification
     * and asynchronous error emitted by the engine.
     */
    addNotificationOutput(output : NotificationDelegate) {
        this._notificationOutputs.add(output);
    }
    /**
     * Remove a previously registered notification API client.
     */
    removeNotificationOutput(output : NotificationDelegate) {
        this._notificationOutputs.delete(output);
    }

    /**
     * Get the list of notification backends that can be used.
     */
    getAvailableNotificationBackends() : Tp.Capabilities.NotificationBackend[] {
        return Object.values(this._staticNotificationBackends)
            .concat(Array.from(this._dynamicNotificationBackends.values()));
    }

    /**
     * Dispatch a notification (a single new result from a stream) from a
     * ThingTalk program.
     *
     * @param app - the running ThingTalk program that generated the notification
     * @param outputType - a string identifying the type of result to display
     * @param outputValue - the new value to display
     */
    async notify(app : AppExecutor, outputType : string, outputValue : Record<string, unknown>) {
        const prefs = this._engine.platform.getSharedPreferences();
        const notificationBackend =
            app.notifications ? app.notifications.backend : (prefs.get('notification-backend') as string || 'conversation');

        const promises = [];
        if (this._notificationOutputs.size > 0 || notificationBackend !== 'conversation') {
            const messages = await this._notificationFormatter.formatNotification(app.name, app.program, outputType, outputValue);

            const notification = {
                appId: app.uniqueId,
                icon: app.icon,
                raw: outputValue,
                type: outputType,
                formatted: messages
            };
            for (const out of this._notificationOutputs.values())
                promises.push(out.notify(notification));

            if (notificationBackend !== 'conversation') {
                if (notificationBackend.startsWith('thingpedia/')) {
                    const deviceId = notificationBackend.substring('thingpedia/'.length);
                    const backend = this._dynamicNotificationBackends.getById(deviceId);
                    if (backend)
                        promises.push(backend.notify(notification));
                } else {
                    const backend = this._staticNotificationBackends[notificationBackend];
                    if (backend)
                        promises.push(backend.notify(notification, app.notifications?.config));
                }
            }
        }

        if (notificationBackend === 'conversation') {
            for (const conv of this._conversations.values())
                promises.push(conv.notify(app, outputType, outputValue));
        }
        await Promise.all(promises);
    }

    async notifyError(app : AppExecutor, error : Error) {
        const prefs = this._engine.platform.getSharedPreferences();
        const notificationBackend =
            app.notifications ? app.notifications.backend : (prefs.get('notification-backend') as string || 'conversation');

        const promises = [];
        const notification = {
            appId: app.uniqueId,
            icon: app.icon,
            error: error
        };
        for (const out of this._notificationOutputs.values())
            promises.push(out.notifyError(notification));

        if (notificationBackend === 'conversation') {
            for (const conv of this._conversations.values())
                promises.push(conv.notifyError(app, error));
        } else if (notificationBackend.startsWith('thingpedia/')) {
            const deviceId = notificationBackend.substring('thingpedia/'.length);
            const backend = this._dynamicNotificationBackends.getById(deviceId);
            if (backend)
                promises.push(backend.notifyError(notification));
        } else {
            const backend = this._staticNotificationBackends[notificationBackend];
            if (backend)
                promises.push(backend.notifyError(notification, app.notifications?.config));
        }

        await Promise.all(promises);
    }

    getConversation(id : string) : Conversation|undefined {
        return this._conversations.get(id);
    }

    get lastConversation() {
        return this._lastConversation;
    }

    pingAll() {
        for (const [_, conversation] of this._conversations)
            conversation.sendPing();
    }


    async getOrOpenConversation(id : string, options : ConversationOptions, state ?: ConversationState) {
        if (this._conversations.has(id))
            return this._conversations.get(id)!;
        options = options || {};
        if (!options.nluServerUrl)
            options.nluServerUrl = this._nluModelUrl;
        const conv = this.openConversation(id, options);
        await conv.start(state);
        return conv;
    }

    openConversation(id : string, options : ConversationOptions) {
        this._conversations.delete(id);
        options = options || {};
        if (!options.nluServerUrl)
            options.nluServerUrl = this._nluModelUrl;

        let deleteWhenInactive = options.deleteWhenInactive;
        if (deleteWhenInactive === undefined)
            deleteWhenInactive = true;
        const conv = new Conversation(this._engine, id, options);
        conv.on('active', () => {
            this._lastConversation = conv;
        });
        if (deleteWhenInactive) {
            conv.on('inactive', () => {
                if (this._conversations.get(conv.id) === conv)
                    this._conversations.delete(conv.id);
            });
        }
        this._lastConversation = conv;
        this._conversations.set(id, conv);
        return conv;
    }

    closeConversation(id : string) {
        this._conversations.delete(id);
    }
}

1	// -- mode: typescript; 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	import assert from 'assert';	97✔
22	import * as Tp from 'thingpedia';
23	import * as events from 'events';	97✔
24
25	import Conversation, {	97✔
26	ConversationDelegate,
27	ConversationOptions,
28	ConversationState
29	} from './conversation';
30	import { PlatformData, Message } from './protocol';
31	import NotificationFormatter from './notifications/formatter';	97✔
32	import {	97✔
33	StaticNotificationBackends,
34	ThingpediaNotificationBackend,
35	NotificationConfig,
36	NotificationDelegate,
37	} from './notifications';
38
39	import AppExecutor from '../engine/apps/app_executor';
40	import type Engine from '../engine';
41	import DeviceView from '../engine/devices/device_view';	97✔
42	import DeviceInterfaceMapper from '../engine/devices/device_interface_mapper';	97✔
43	import { ConversationStateRow, LocalTable } from "../engine/db";
44
45	/**
46	* A conversation delegate that buffers all commands until the dialogue turn
47	* is complete.
48	*/
49	class StatelessConversationDelegate implements ConversationDelegate {
50	private _buffer : Message[];
51	private _askSpecial : string\|null;
52	private _id : string;
53
54	constructor(conversationId : string) {
55	this._buffer = [];	×
56	this._askSpecial = null;	×
57	this._id = conversationId;	×
58	}
59
60	flush() {
61	const buffer = this._buffer;	×
62	const askSpecial = this._askSpecial;	×
63	this._buffer = [];	×
64	this._askSpecial = null;	×
65	return {	×
66	conversationId : this._id,
67	messages: buffer,
68	askSpecial: askSpecial,
69	};
70	}
71
72	setHypothesis() {
73	// ignore
74	}
75
76	setExpected(what : string\|null) {
77	assert(this._askSpecial === null);	×
78	this._askSpecial = what;	×
79	}
80
81	async addMessage(message : Message) {
82	this._buffer.push(message);	×
83	}
84	}
85
86	export interface CommandInput {
87	type : 'command';
88	text : string;
89	}
90
91	export interface ParsedInput {
92	type : 'parsed';
93	json : any;
94	title ?: string;
95	}
96
97	export interface ThingTalkInput {
98	type : 'tt';
99	code : string;
100	}
101
102	export type ConverseInput = (CommandInput \| ParsedInput \| ThingTalkInput) & PlatformData;
103
104	/**
105	* The main controller class for interaction with the user.
106	*
107	* This class manages multiple conversations (independent sessions of the
108	* agent with their own state), it dispatches notifications, and it
109	* handles connections from API clients.
110	*/
111	export default class AssistantDispatcher extends events.EventEmitter {	97✔
112	private _engine : Engine;
113	private _notificationFormatter : NotificationFormatter;
114	private _nluModelUrl : string\|undefined;
115
116	private _notificationOutputs : Set<NotificationDelegate>;
117	private _staticNotificationBackends : Record<string, Tp.Capabilities.NotificationBackend>;
118	private _dynamicNotificationBackends : DeviceInterfaceMapper<Tp.Capabilities.NotificationBackend>;
119	private _conversations : Map<string, Conversation>;
120	private _lastConversation : Conversation\|null;
121	private _conversationStateDB : LocalTable<ConversationStateRow>;
122
123	constructor(engine : Engine, nluModelUrl : string\|undefined, notificationConfig : NotificationConfig) {
124	super();	5✔
125
126	this._engine = engine;	5✔
127	this._notificationFormatter = new NotificationFormatter(engine);	5✔
128	this._nluModelUrl = nluModelUrl;	5✔
129	this._notificationOutputs = new Set;	5✔
130	this._conversations = new Map;	5✔
131	this._lastConversation = null;	5✔
132	this._conversationStateDB = this._engine.db.getLocalTable('conversation_state');	5✔
133
134	this._dynamicNotificationBackends = new DeviceInterfaceMapper(	5✔
135	new DeviceView(engine.devices, 'org.thingpedia.notification-provider', {}),
136	(device) => new ThingpediaNotificationBackend(device));	×
137
138	// initialize static notification backends
139	this._staticNotificationBackends = {};	5✔
140	for (const key in notificationConfig) {	5✔
141	const key2 = key as keyof typeof StaticNotificationBackends;	×
142	this._staticNotificationBackends[key] = new (StaticNotificationBackends[key2])(engine, notificationConfig[key2] as any);	×
143	}
144	}
145
146	async start() {
147	await this._notificationFormatter.initialize();	5✔
148	this._dynamicNotificationBackends.start();	5✔
149	}
150	async stop() {
151	this._dynamicNotificationBackends.stop();	5✔
152	}
153
154	/**
155	* Dispatch one single command to an existing conversation.
156	*
157	* This is an alternative to getting the conversation and adding a
158	* ConversationDelegate to it.
159	* It exists for the convenience of REST API clients which do not keep
160	* an open web socket.
161	*/
162	async converse(command : ConverseInput, conversationId : string) : Promise<{
163	conversationId : string;
164	messages : Message[],
165	askSpecial : string\|null;
166	}> {
NEW 167	const state = await this._conversationStateDB.getOne(conversationId).then((row) => {	×
NEW 168	if (row) {	×
NEW 169	return {	×
170	history : row.history ? JSON.parse(row.history) : [],	×
171	dialogueState : row.dialogueState ? JSON.parse(row.dialogueState) : null,	×
172	lastMessageId : row.lastMessageId ? row.lastMessageId : 0	×
173	};
174	}
NEW 175	return undefined;	×
176	});
177
UNCOV 178	const conversation = await this.getOrOpenConversation(conversationId, {	×
179	showWelcome: false,
180	anonymous: false,
181	debug: true
182	}, state);
183	const delegate = new StatelessConversationDelegate(conversationId);	×
184	await conversation.addOutput(delegate, false);	×
185
186	switch (command.type) {	×
187	case 'command':	×
188	await conversation.handleCommand(command.text, command);	×
189	break;	×
190	case 'parsed':
191	await conversation.handleParsedCommand(command.json, command.title, command);	×
192	break;	×
193	case 'tt':
194	await conversation.handleThingTalk(command.code, command);	×
195	break;	×
196	}
197
198	await conversation.removeOutput(delegate);	×
199	const result = delegate.flush();	×
200	return result;	×
201	}
202
203	/**
204	* Add an API client that wishes to listen for notifications.
205	*
206	* The delegate methods will be invoked for every notification
207	* and asynchronous error emitted by the engine.
208	*/
209	addNotificationOutput(output : NotificationDelegate) {
210	this._notificationOutputs.add(output);	11✔
211	}
212	/**
213	* Remove a previously registered notification API client.
214	*/
215	removeNotificationOutput(output : NotificationDelegate) {
216	this._notificationOutputs.delete(output);	11✔
217	}
218
219	/**
220	* Get the list of notification backends that can be used.
221	*/
222	getAvailableNotificationBackends() : Tp.Capabilities.NotificationBackend[] {
223	return Object.values(this._staticNotificationBackends)	×
224	.concat(Array.from(this._dynamicNotificationBackends.values()));
225	}
226
227	/**
228	* Dispatch a notification (a single new result from a stream) from a
229	* ThingTalk program.
230	*
231	* @param app - the running ThingTalk program that generated the notification
232	* @param outputType - a string identifying the type of result to display
233	* @param outputValue - the new value to display
234	*/
235	async notify(app : AppExecutor, outputType : string, outputValue : Record<string, unknown>) {
236	const prefs = this._engine.platform.getSharedPreferences();	25✔
237	const notificationBackend =
238	app.notifications ? app.notifications.backend : (prefs.get('notification-backend') as string \|\| 'conversation');	25!
239
240	const promises = [];	25✔
241	if (this._notificationOutputs.size > 0 \|\| notificationBackend !== 'conversation') {	25!
242	const messages = await this._notificationFormatter.formatNotification(app.name, app.program, outputType, outputValue);	25✔
243
244	const notification = {	25✔
245	appId: app.uniqueId,
246	icon: app.icon,
247	raw: outputValue,
248	type: outputType,
249	formatted: messages
250	};
251	for (const out of this._notificationOutputs.values())	25✔
252	promises.push(out.notify(notification));	25✔
253
254	if (notificationBackend !== 'conversation') {	25!
255	if (notificationBackend.startsWith('thingpedia/')) {	×
256	const deviceId = notificationBackend.substring('thingpedia/'.length);	×
257	const backend = this._dynamicNotificationBackends.getById(deviceId);	×
258	if (backend)	×
259	promises.push(backend.notify(notification));	×
260	} else {
261	const backend = this._staticNotificationBackends[notificationBackend];	×
262	if (backend)	×
263	promises.push(backend.notify(notification, app.notifications?.config));	×
264	}
265	}
266	}
267
268	if (notificationBackend === 'conversation') {	25!
269	for (const conv of this._conversations.values())	25✔
270	promises.push(conv.notify(app, outputType, outputValue));	25✔
271	}
272	await Promise.all(promises);	25✔
273	}
274
275	async notifyError(app : AppExecutor, error : Error) {
276	const prefs = this._engine.platform.getSharedPreferences();	3✔
277	const notificationBackend =
278	app.notifications ? app.notifications.backend : (prefs.get('notification-backend') as string \|\| 'conversation');	3!
279
280	const promises = [];	3✔
281	const notification = {	3✔
282	appId: app.uniqueId,
283	icon: app.icon,
284	error: error
285	};
286	for (const out of this._notificationOutputs.values())	3✔
287	promises.push(out.notifyError(notification));	3✔
288
289	if (notificationBackend === 'conversation') {	3!
290	for (const conv of this._conversations.values())	3✔
291	promises.push(conv.notifyError(app, error));	3✔
292	} else if (notificationBackend.startsWith('thingpedia/')) {	×
293	const deviceId = notificationBackend.substring('thingpedia/'.length);	×
294	const backend = this._dynamicNotificationBackends.getById(deviceId);	×
295	if (backend)	×
296	promises.push(backend.notifyError(notification));	×
297	} else {
298	const backend = this._staticNotificationBackends[notificationBackend];	×
299	if (backend)	×
300	promises.push(backend.notifyError(notification, app.notifications?.config));	×
301	}
302
303	await Promise.all(promises);	3✔
304	}
305
306	getConversation(id : string) : Conversation\|undefined {
307	return this._conversations.get(id);	×
308	}
309
310	get lastConversation() {
311	return this._lastConversation;	×
312	}
313
314	pingAll() {
315	for (const [_, conversation] of this._conversations)	8✔
316	conversation.sendPing();	8✔
317	}
318
319
320	async getOrOpenConversation(id : string, options : ConversationOptions, state ?: ConversationState) {
321	if (this._conversations.has(id))	2!
322	return this._conversations.get(id)!;	×
323	options = options \|\| {};	2!
324	if (!options.nluServerUrl)	2!
325	options.nluServerUrl = this._nluModelUrl;	2✔
326	const conv = this.openConversation(id, options);	2✔
327	await conv.start(state);	2✔
328	return conv;	2✔
329	}
330
331	openConversation(id : string, options : ConversationOptions) {
332	this._conversations.delete(id);	3✔
333	options = options \|\| {};	3✔
334	if (!options.nluServerUrl)	3✔
335	options.nluServerUrl = this._nluModelUrl;	1✔
336
337	let deleteWhenInactive = options.deleteWhenInactive;	3✔
338	if (deleteWhenInactive === undefined)	3!
339	deleteWhenInactive = true;	3✔
340	const conv = new Conversation(this._engine, id, options);	3✔
341	conv.on('active', () => {	3✔
342	this._lastConversation = conv;	2✔
343	});
344	if (deleteWhenInactive) {	3!
345	conv.on('inactive', () => {	3✔
346	if (this._conversations.get(conv.id) === conv)	×
347	this._conversations.delete(conv.id);	×
348	});
349	}
350	this._lastConversation = conv;	3✔
351	this._conversations.set(id, conv);	3✔
352	return conv;	3✔
353	}
354
355	closeConversation(id : string) {
356	this._conversations.delete(id);	×
357	}
358	}

stanford-oval / genie-toolkit / 2260

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