From 2f4f7cfd0d19017f296214d8c676990e13072efe Mon Sep 17 00:00:00 2001 From: Najim Date: Thu, 8 Feb 2024 14:36:16 +0100 Subject: [PATCH] fix(server): make leon handle multiple socket.io-client instances --- server/src/core/asr/asr.ts | 6 +- server/src/core/brain/brain.ts | 55 +++++++--- .../core/http-server/api/utterance/post.ts | 3 +- server/src/core/index.ts | 7 +- server/src/core/nlp/nlu/nlu.ts | 63 ++++++----- server/src/core/nlp/nlu/slot-filling.ts | 94 ++++++++-------- server/src/core/socket-server.ts | 102 ++++++++++-------- server/src/core/stt/stt.ts | 15 ++- .../synthesizers/amazon-polly-synthesizer.ts | 12 ++- .../tts/synthesizers/flite-synthesizer.ts | 12 ++- .../google-cloud-tts-synthesizer.ts | 16 ++- .../synthesizers/watson-tts-synthesizer.ts | 11 +- server/src/core/tts/tts.ts | 11 +- 13 files changed, 247 insertions(+), 160 deletions(-) diff --git a/server/src/core/asr/asr.ts b/server/src/core/asr/asr.ts index dfa02a8454..da6215552e 100644 --- a/server/src/core/asr/asr.ts +++ b/server/src/core/asr/asr.ts @@ -8,6 +8,8 @@ import { TMP_PATH } from '@/constants' import { STT } from '@/core' import { LogHelper } from '@/helpers/log-helper' +import { ClientSocket } from '../socket-server' + export default class ASR { private static instance: ASR @@ -29,7 +31,7 @@ export default class ASR { * Encode audio blob to WAVE file * and forward the WAVE file to the STT parser */ - public encode(blob: Buffer): Promise { + public encode(blob: Buffer, socket: ClientSocket): Promise { return new Promise((resolve, reject) => { LogHelper.title('ASR') @@ -60,7 +62,7 @@ export default class ASR { if (!STT.isParserReady) { reject(new Error('The speech recognition is not ready yet')) } else { - STT.transcribe(this.audioPaths.wav) + STT.transcribe(this.audioPaths.wav, socket) resolve() } }) diff --git a/server/src/core/brain/brain.ts b/server/src/core/brain/brain.ts index 68ce00cf1d..8bff19508c 100644 --- a/server/src/core/brain/brain.ts +++ b/server/src/core/brain/brain.ts @@ -4,6 +4,7 @@ import { ChildProcessWithoutNullStreams, spawn } from 'node:child_process' import type { ShortLanguageCode } from '@/types' import type { GlobalAnswersSchema } from '@/schemas/global-data-schemas' +import TextToSpeech from '@/core/tts/tts' import type { CustomEnumEntity, NERCustomEntity, @@ -27,7 +28,6 @@ import { NODEJS_BRIDGE_BIN_PATH, TMP_PATH } from '@/constants' -import { SOCKET_SERVER, TTS } from '@/core' import { LangHelper } from '@/helpers/lang-helper' import { LogHelper } from '@/helpers/log-helper' import { SkillDomainHelper } from '@/helpers/skill-domain-helper' @@ -35,6 +35,8 @@ import { StringHelper } from '@/helpers/string-helper' import type { AnswerOutput } from '@sdk/types' import { DateHelper } from '@/helpers/date-helper' +import { ClientSocket } from '../socket-server' + export default class Brain { private static instance: Brain private _lang: ShortLanguageCode = 'en' @@ -49,15 +51,29 @@ export default class Brain { private skillFriendlyName = '' private skillOutput = '' private answers: SkillAnswerConfigSchema[] = [] - public isMuted = false // Close Leon mouth if true; e.g. over HTTP - constructor() { + private _socket: ClientSocket | undefined + public get socket(): ClientSocket | undefined { + return this._socket + } + public get isMuted(): boolean { + return this._socket == undefined + } + + private _tts: TextToSpeech + public get tts(): TextToSpeech { + return this._tts + } + + constructor(socket?: ClientSocket) { if (!Brain.instance) { LogHelper.title('Brain') LogHelper.success('New instance') Brain.instance = this } + this._socket = socket + this._tts = new TextToSpeech(this) } public get lang(): ShortLanguageCode { @@ -80,7 +96,7 @@ export default class Brain { } private async updateTTSLang(newLang: ShortLanguageCode): Promise { - await TTS.init(newLang) + await this.tts.init(newLang) LogHelper.title('Brain') LogHelper.info('Language has changed') @@ -99,6 +115,15 @@ export default class Brain { } } + /** + * Resources free-up. + */ + public dispose(): void { + this._socket?.disconnect() + this._socket?.removeAllListeners() + this._socket = undefined + } + /** * Make Leon talk */ @@ -114,10 +139,10 @@ export default class Brain { // Stripe HTML to a whitespace. Whitespace to let the TTS respects punctuation const speech = speechAnswer.replace(/<(?:.|\n)*?>/gm, ' ') - TTS.add(speech, end) + this.tts.add(speech, end) } - SOCKET_SERVER.socket?.emit('answer', textAnswer) + this.socket?.emit('answer', textAnswer) } } @@ -166,7 +191,7 @@ export default class Brain { const speech = `${this.wernicke('random_not_sure')}.` this.talk(speech, true) - SOCKET_SERVER.socket?.emit('ask-to-repeat', nluResult) + this.socket?.emit('ask-to-repeat', nluResult) } } @@ -219,7 +244,7 @@ export default class Brain { LogHelper.info(data.toString()) if (skillAnswer.output.widget) { - SOCKET_SERVER.socket?.emit('widget', skillAnswer.output.widget) + this.socket?.emit('widget', skillAnswer.output.widget) } const { answer } = skillAnswer.output @@ -254,7 +279,7 @@ export default class Brain { if (!this.isMuted) { this.talk(speech) - SOCKET_SERVER.socket?.emit('is-typing', false) + this.socket?.emit('is-typing', false) } this.answers.push(speech) @@ -428,9 +453,7 @@ export default class Brain { Brain.deleteIntentObjFile(intentObjectPath) - if (!this.isMuted) { - SOCKET_SERVER.socket?.emit('is-typing', false) - } + this.socket?.emit('is-typing', false) const executionTimeEnd = Date.now() const executionTime = executionTimeEnd - executionTimeStart @@ -440,13 +463,13 @@ export default class Brain { nextAction?.suggestions && skillResult?.output.core?.showNextActionSuggestions ) { - SOCKET_SERVER.socket?.emit('suggest', nextAction.suggestions) + this.socket?.emit('suggest', nextAction.suggestions) } if ( action?.suggestions && skillResult?.output.core?.showSuggestions ) { - SOCKET_SERVER.socket?.emit('suggest', action.suggestions) + this.socket?.emit('suggest', action.suggestions) } resolve({ @@ -574,12 +597,12 @@ export default class Brain { if (!this.isMuted) { this.talk(answer as string, true) - SOCKET_SERVER.socket?.emit('is-typing', false) + this.socket?.emit('is-typing', false) } // Send suggestions to the client if (nextAction?.suggestions) { - SOCKET_SERVER.socket?.emit('suggest', nextAction.suggestions) + this.socket?.emit('suggest', nextAction.suggestions) } resolve({ diff --git a/server/src/core/http-server/api/utterance/post.ts b/server/src/core/http-server/api/utterance/post.ts index 7df453e468..9220690f90 100644 --- a/server/src/core/http-server/api/utterance/post.ts +++ b/server/src/core/http-server/api/utterance/post.ts @@ -2,7 +2,7 @@ import type { FastifyPluginAsync, FastifySchema } from 'fastify' import { Type } from '@sinclair/typebox' import type { Static } from '@sinclair/typebox' -import { NLU, BRAIN } from '@/core' +import { NLU } from '@/core' import type { APIOptions } from '@/core/http-server/http-server' const postUtteranceSchema = { @@ -29,7 +29,6 @@ export const postUtterance: FastifyPluginAsync = async ( const { utterance } = request.body try { - BRAIN.isMuted = true const data = await NLU.process(utterance) reply.send({ diff --git a/server/src/core/index.ts b/server/src/core/index.ts index 2c4edf15bf..918332305e 100644 --- a/server/src/core/index.ts +++ b/server/src/core/index.ts @@ -3,7 +3,6 @@ import TCPClient from '@/core/tcp-client' import HTTPServer from '@/core/http-server/http-server' import SocketServer from '@/core/socket-server' import SpeechToText from '@/core/stt/stt' -import TextToSpeech from '@/core/tts/tts' import AutomaticSpeechRecognition from '@/core/asr/asr' import NamedEntityRecognition from '@/core/nlp/nlu/ner' import ModelLoader from '@/core/nlp/nlu/model-loader' @@ -25,14 +24,12 @@ export const SOCKET_SERVER = new SocketServer() export const STT = new SpeechToText() -export const TTS = new TextToSpeech() - export const ASR = new AutomaticSpeechRecognition() export const NER = new NamedEntityRecognition() export const MODEL_LOADER = new ModelLoader() -export const NLU = new NaturalLanguageUnderstanding() - +// TODO: THIS IS USED OVER HTTP (FOR HTTP REQUESTS) export const BRAIN = new Brain() +export const NLU = new NaturalLanguageUnderstanding(BRAIN) diff --git a/server/src/core/nlp/nlu/nlu.ts b/server/src/core/nlp/nlu/nlu.ts index 2d664d62d8..983be67256 100644 --- a/server/src/core/nlp/nlu/nlu.ts +++ b/server/src/core/nlp/nlu/nlu.ts @@ -15,7 +15,8 @@ import type { } from '@/core/nlp/types' import { langs } from '@@/core/langs.json' import { TCP_SERVER_BIN_PATH } from '@/constants' -import { TCP_CLIENT, BRAIN, SOCKET_SERVER, MODEL_LOADER, NER } from '@/core' +import { TCP_CLIENT, MODEL_LOADER, NER } from '@/core' +import Brain from '@/core/brain/brain' import { LogHelper } from '@/helpers/log-helper' import { LangHelper } from '@/helpers/lang-helper' import { ActionLoop } from '@/core/nlp/nlu/action-loop' @@ -45,14 +46,22 @@ export default class NLU { private static instance: NLU public nluResult: NLUResult = DEFAULT_NLU_RESULT public conversation = new Conversation('conv0') + private slotFilling!: SlotFilling - constructor() { + private _brain: Brain + public get brain(): Brain { + return this._brain + } + + constructor(brain: Brain) { if (!NLU.instance) { LogHelper.title('NLU') LogHelper.success('New instance') NLU.instance = this } + this._brain = brain + this.slotFilling = new SlotFilling(this) } /** @@ -66,8 +75,8 @@ export default class NLU { await this.process(utterance) } - BRAIN.lang = locale - BRAIN.talk(`${BRAIN.wernicke('random_language_switch')}.`, true) + this.brain.lang = locale + this.brain.talk(`${this.brain.wernicke('random_language_switch')}.`, true) // Recreate a new TCP server process and reconnect the TCP client kill(global.tcpServerProcess.pid as number, () => { @@ -94,9 +103,9 @@ export default class NLU { LogHelper.info('Processing...') if (!MODEL_LOADER.hasNlpModels()) { - if (!BRAIN.isMuted) { - BRAIN.talk(`${BRAIN.wernicke('random_errors')}!`) - SOCKET_SERVER.socket?.emit('is-typing', false) + if (this.brain.socket !== undefined) { + this.brain.talk(`${this.brain.wernicke('random_errors')}!`) + this.brain.socket?.emit('is-typing', false) } const msg = @@ -118,7 +127,7 @@ export default class NLU { // When the active context has slots filled if (Object.keys(this.conversation.activeContext.slots).length > 0) { try { - return resolve(await SlotFilling.handle(utterance)) + return resolve(await this.slotFilling.handle(utterance)) } catch (e) { return reject({}) } @@ -174,13 +183,16 @@ export default class NLU { const isSupportedLanguage = LangHelper.getShortCodes().includes(locale) if (!isSupportedLanguage) { - BRAIN.talk(`${BRAIN.wernicke('random_language_not_supported')}.`, true) - SOCKET_SERVER.socket?.emit('is-typing', false) + this.brain.talk( + `${this.brain.wernicke('random_language_not_supported')}.`, + true + ) + this.brain.socket?.emit('is-typing', false) return resolve({}) } // Trigger language switching - if (BRAIN.lang !== locale) { + if (this.brain.lang !== locale) { this.switchLanguage(utterance, locale) return resolve(null) } @@ -191,16 +203,19 @@ export default class NLU { ) if (!fallback) { - if (!BRAIN.isMuted) { - BRAIN.talk(`${BRAIN.wernicke('random_unknown_intents')}.`, true) - SOCKET_SERVER.socket?.emit('is-typing', false) + if (this.brain.socket !== undefined) { + this.brain.talk( + `${this.brain.wernicke('random_unknown_intents')}.`, + true + ) + this.brain.socket?.emit('is-typing', false) } LogHelper.title('NLU') const msg = 'Intent not found' LogHelper.warning(msg) - Telemetry.utterance({ utterance, lang: BRAIN.lang }) + Telemetry.utterance({ utterance, lang: this.brain.lang }) return resolve(null) } @@ -219,13 +234,13 @@ export default class NLU { this.nluResult.classification.domain, this.nluResult.classification.skill, 'config', - BRAIN.lang + '.json' + this.brain.lang + '.json' ) this.nluResult.skillConfigPath = skillConfigPath try { this.nluResult.entities = await NER.extractEntities( - BRAIN.lang, + this.brain.lang, skillConfigPath, this.nluResult ) @@ -233,7 +248,7 @@ export default class NLU { LogHelper.error(`Failed to extract entities: ${e}`) } - const shouldSlotLoop = await SlotFilling.route(intent) + const shouldSlotLoop = await this.slotFilling.route(intent) if (shouldSlotLoop) { return resolve({}) } @@ -244,7 +259,7 @@ export default class NLU { Object.keys(this.conversation.activeContext.slots).length > 0 ) { try { - return resolve(await SlotFilling.handle(utterance)) + return resolve(await this.slotFilling.handle(utterance)) } catch (e) { return reject({}) } @@ -256,7 +271,7 @@ export default class NLU { } await this.conversation.setActiveContext({ ...DEFAULT_ACTIVE_CONTEXT, - lang: BRAIN.lang, + lang: this.brain.lang, slots: {}, isInActionLoop: false, originalUtterance: this.nluResult.utterance, @@ -273,14 +288,14 @@ export default class NLU { this.nluResult.entities = this.conversation.activeContext.entities try { - const processedData = await BRAIN.execute(this.nluResult) + const processedData = await this.brain.execute(this.nluResult) // Prepare next action if there is one queuing if (processedData.nextAction) { this.conversation.cleanActiveContext() await this.conversation.setActiveContext({ ...DEFAULT_ACTIVE_CONTEXT, - lang: BRAIN.lang, + lang: this.brain.lang, slots: {}, isInActionLoop: !!processedData.nextAction.loop, originalUtterance: processedData.utterance ?? '', @@ -306,9 +321,7 @@ export default class NLU { LogHelper.error(errorMessage) - if (!BRAIN.isMuted) { - SOCKET_SERVER.socket?.emit('is-typing', false) - } + this.brain.socket?.emit('is-typing', false) return reject(new Error(errorMessage)) } diff --git a/server/src/core/nlp/nlu/slot-filling.ts b/server/src/core/nlp/nlu/slot-filling.ts index f4a41e2a9e..5d9c1925a0 100644 --- a/server/src/core/nlp/nlu/slot-filling.ts +++ b/server/src/core/nlp/nlu/slot-filling.ts @@ -2,16 +2,22 @@ import { join } from 'node:path' import type { NLPUtterance } from '@/core/nlp/types' import type { BrainProcessResult } from '@/core/brain/types' -import { BRAIN, MODEL_LOADER, NER, NLU, SOCKET_SERVER } from '@/core' -import { DEFAULT_NLU_RESULT } from '@/core/nlp/nlu/nlu' +import { MODEL_LOADER, NER } from '@/core' +import NaturalLanguageUnderstanding, { + DEFAULT_NLU_RESULT +} from '@/core/nlp/nlu/nlu' import { SkillDomainHelper } from '@/helpers/skill-domain-helper' import { DEFAULT_ACTIVE_CONTEXT } from '@/core/nlp/conversation' export class SlotFilling { + private nlu!: NaturalLanguageUnderstanding + constructor(nlu: NaturalLanguageUnderstanding) { + this.nlu = nlu + } /** * Handle slot filling */ - public static async handle( + public async handle( utterance: NLPUtterance ): Promise | null> { const processedData = await this.fillSlot(utterance) @@ -21,16 +27,16 @@ export class SlotFilling { * Then reprocess with the new utterance */ if (!processedData) { - await NLU.process(utterance) + await this.nlu.process(utterance) return null } if (processedData && Object.keys(processedData).length > 0) { // Set new context with the next action if there is one if (processedData.action?.next_action) { - await NLU.conversation.setActiveContext({ + await this.nlu.conversation.setActiveContext({ ...DEFAULT_ACTIVE_CONTEXT, - lang: BRAIN.lang, + lang: this.nlu.brain.lang, slots: processedData.slots || {}, isInActionLoop: !!processedData.nextAction?.loop, originalUtterance: processedData.utterance ?? null, @@ -50,14 +56,14 @@ export class SlotFilling { * Build NLU data result object based on slots * and ask for more entities if necessary */ - public static async fillSlot( + public async fillSlot( utterance: NLPUtterance ): Promise | null> { - if (!NLU.conversation.activeContext.nextAction) { + if (!this.nlu.conversation.activeContext.nextAction) { return null } - const { domain, intent } = NLU.conversation.activeContext + const { domain, intent } = this.nlu.conversation.activeContext const [skillName, actionName] = intent.split('.') as [string, string] const skillConfigPath = join( process.cwd(), @@ -65,10 +71,10 @@ export class SlotFilling { domain, skillName, 'config', - BRAIN.lang + '.json' + this.nlu.brain.lang + '.json' ) - NLU.nluResult = { + this.nlu.nluResult = { ...DEFAULT_NLU_RESULT, // Reset entities, slots, etc. utterance, classification: { @@ -80,56 +86,58 @@ export class SlotFilling { } const entities = await NER.extractEntities( - BRAIN.lang, + this.nlu.brain.lang, skillConfigPath, - NLU.nluResult + this.nlu.nluResult ) // Continue to loop for questions if a slot has been filled correctly - let notFilledSlot = NLU.conversation.getNotFilledSlot() + let notFilledSlot = this.nlu.conversation.getNotFilledSlot() if (notFilledSlot && entities.length > 0) { const hasMatch = entities.some( ({ entity }) => entity === notFilledSlot?.expectedEntity ) if (hasMatch) { - NLU.conversation.setSlots(BRAIN.lang, entities) + this.nlu.conversation.setSlots(this.nlu.brain.lang, entities) - notFilledSlot = NLU.conversation.getNotFilledSlot() + notFilledSlot = this.nlu.conversation.getNotFilledSlot() if (notFilledSlot) { - BRAIN.talk(notFilledSlot.pickedQuestion) - SOCKET_SERVER.socket?.emit('is-typing', false) + this.nlu.brain.talk(notFilledSlot.pickedQuestion) + this.nlu.brain.socket?.emit('is-typing', false) return {} } } } - if (!NLU.conversation.areSlotsAllFilled()) { - BRAIN.talk(`${BRAIN.wernicke('random_context_out_of_topic')}.`) + if (!this.nlu.conversation.areSlotsAllFilled()) { + this.nlu.brain.talk( + `${this.nlu.brain.wernicke('random_context_out_of_topic')}.` + ) } else { - NLU.nluResult = { + this.nlu.nluResult = { ...DEFAULT_NLU_RESULT, // Reset entities, slots, etc. // Assign slots only if there is a next action - slots: NLU.conversation.activeContext.nextAction - ? NLU.conversation.activeContext.slots + slots: this.nlu.conversation.activeContext.nextAction + ? this.nlu.conversation.activeContext.slots : {}, - utterance: NLU.conversation.activeContext.originalUtterance ?? '', + utterance: this.nlu.conversation.activeContext.originalUtterance ?? '', skillConfigPath, classification: { domain, skill: skillName, - action: NLU.conversation.activeContext.nextAction, + action: this.nlu.conversation.activeContext.nextAction, confidence: 1 } } - NLU.conversation.cleanActiveContext() + this.nlu.conversation.cleanActiveContext() - return BRAIN.execute(NLU.nluResult) + return this.nlu.brain.execute(this.nlu.nluResult) } - NLU.conversation.cleanActiveContext() + this.nlu.conversation.cleanActiveContext() return null } @@ -139,40 +147,40 @@ export class SlotFilling { * 2. If the context is expecting slots, then loop over questions to slot fill * 3. Or go to the brain executor if all slots have been filled in one shot */ - public static async route(intent: string): Promise { + public async route(intent: string): Promise { const slots = await MODEL_LOADER.mainNLPContainer.slotManager.getMandatorySlots(intent) const hasMandatorySlots = Object.keys(slots)?.length > 0 if (hasMandatorySlots) { - await NLU.conversation.setActiveContext({ + await this.nlu.conversation.setActiveContext({ ...DEFAULT_ACTIVE_CONTEXT, - lang: BRAIN.lang, + lang: this.nlu.brain.lang, slots, isInActionLoop: false, - originalUtterance: NLU.nluResult.utterance, - skillConfigPath: NLU.nluResult.skillConfigPath, - actionName: NLU.nluResult.classification.action, - domain: NLU.nluResult.classification.domain, + originalUtterance: this.nlu.nluResult.utterance, + skillConfigPath: this.nlu.nluResult.skillConfigPath, + actionName: this.nlu.nluResult.classification.action, + domain: this.nlu.nluResult.classification.domain, intent, - entities: NLU.nluResult.entities + entities: this.nlu.nluResult.entities }) - const notFilledSlot = NLU.conversation.getNotFilledSlot() + const notFilledSlot = this.nlu.conversation.getNotFilledSlot() // Loop for questions if a slot hasn't been filled if (notFilledSlot) { const { actions } = await SkillDomainHelper.getSkillConfig( - NLU.nluResult.skillConfigPath, - BRAIN.lang + this.nlu.nluResult.skillConfigPath, + this.nlu.brain.lang ) const [currentSlot] = - actions[NLU.nluResult.classification.action]?.slots?.filter( + actions[this.nlu.nluResult.classification.action]?.slots?.filter( ({ name }) => name === notFilledSlot.name ) ?? [] - SOCKET_SERVER.socket?.emit('suggest', currentSlot?.suggestions) - BRAIN.talk(notFilledSlot.pickedQuestion) - SOCKET_SERVER.socket?.emit('is-typing', false) + this.nlu.brain.socket?.emit('suggest', currentSlot?.suggestions) + this.nlu.brain.talk(notFilledSlot.pickedQuestion) + this.nlu.brain.socket?.emit('is-typing', false) return true } diff --git a/server/src/core/socket-server.ts b/server/src/core/socket-server.ts index 919723755f..79fb87aba1 100644 --- a/server/src/core/socket-server.ts +++ b/server/src/core/socket-server.ts @@ -2,19 +2,13 @@ import type { DefaultEventsMap } from 'socket.io/dist/typed-events' import { Server as SocketIOServer, Socket } from 'socket.io' import { LANG, HAS_STT, HAS_TTS, IS_DEVELOPMENT_ENV } from '@/constants' -import { - HTTP_SERVER, - TCP_CLIENT, - ASR, - STT, - TTS, - NLU, - BRAIN, - MODEL_LOADER -} from '@/core' +import { HTTP_SERVER, TCP_CLIENT, ASR, STT, MODEL_LOADER } from '@/core' import { LogHelper } from '@/helpers/log-helper' import { LangHelper } from '@/helpers/lang-helper' import { Telemetry } from '@/telemetry' +import TextToSpeech from '@/core/tts/tts' +import NaturalLanguageUnderstanding from '@/core/nlp/nlu/nlu' +import Brain from '@/core/brain/brain' interface HotwordDataEvent { hotword: string @@ -26,19 +20,47 @@ interface UtteranceDataEvent { value: string } -export default class SocketServer { - private static instance: SocketServer +export type ClientSocket = Socket - public socket: Socket | undefined = - undefined +export default class SocketServer { + private nlus: Map = new Map() + private onConnectionReady!: (socket: ClientSocket) => void + private onDisconnect!: (socket: ClientSocket) => void constructor() { - if (!SocketServer.instance) { - LogHelper.title('Socket Server') - LogHelper.success('New instance') + LogHelper.title('Socket Server') + LogHelper.success('New instance') + ;(this.onConnectionReady = (socket): void => { + const brain = new Brain(socket) + const nlu = new NaturalLanguageUnderstanding(brain) + let ttsState = 'disabled' + this.nlus.set(socket.id, nlu) + if (HAS_TTS) { + this.getSocketTextToSpeech(socket.id) + ?.init(LangHelper.getShortCode(LANG)) + ?.then(() => { + ttsState = 'enabled' + LogHelper.success(`TTS ${ttsState} for client ${socket.id}`) + }) + } + }), + (this.onDisconnect = (socket): void => { + const { id } = socket + this.nlus.get(id)?.brain.dispose() + this.nlus.delete(id) + }) + } - SocketServer.instance = this - } + private getSocketNlu( + socketId: string + ): NaturalLanguageUnderstanding | undefined { + return this.nlus.get(socketId) + } + private getSocketBrain(socketId: string): Brain | undefined { + return this.getSocketNlu(socketId)?.brain + } + private getSocketTextToSpeech(socketId: string): TextToSpeech | undefined { + return this.getSocketBrain(socketId)?.tts } public async init(): Promise { @@ -49,22 +71,15 @@ export default class SocketServer { : new SocketIOServer(HTTP_SERVER.httpServer) let sttState = 'disabled' - let ttsState = 'disabled' if (HAS_STT) { sttState = 'enabled' await STT.init() } - if (HAS_TTS) { - ttsState = 'enabled' - - await TTS.init(LangHelper.getShortCode(LANG)) - } LogHelper.title('Initialization') LogHelper.success(`STT ${sttState}`) - LogHelper.success(`TTS ${ttsState}`) try { await MODEL_LOADER.loadNLPModels() @@ -76,47 +91,45 @@ export default class SocketServer { LogHelper.title('Client') LogHelper.success('Connected') - this.socket = socket - // Init - this.socket.on('init', (data: string) => { + socket.on('init', (data: string) => { LogHelper.info(`Type: ${data}`) - LogHelper.info(`Socket ID: ${this.socket?.id}`) + LogHelper.info(`Socket ID: ${socket.id}`) - // TODO - // const provider = await addProvider(socket.id) + this.onConnectionReady(socket) // Check whether the TCP client is connected to the TCP server if (TCP_CLIENT.isConnected) { - this.socket?.emit('ready') + socket.emit('ready') } else { TCP_CLIENT.ee.on('connected', () => { - this.socket?.emit('ready') + socket.emit('ready') }) } if (data === 'hotword-node') { // Hotword triggered - this.socket?.on('hotword-detected', (data: HotwordDataEvent) => { + socket.on('hotword-detected', (data: HotwordDataEvent) => { LogHelper.title('Socket') LogHelper.success(`Hotword ${data.hotword} detected`) - this.socket?.broadcast.emit('enable-record') + socket.broadcast.emit('enable-record') }) } else { // Listen for new utterance - this.socket?.on('utterance', async (data: UtteranceDataEvent) => { + socket.on('utterance', async (data: UtteranceDataEvent) => { LogHelper.title('Socket') LogHelper.info(`${data.client} emitted: ${data.value}`) - this.socket?.emit('is-typing', true) + socket.emit('is-typing', true) const { value: utterance } = data try { LogHelper.time('Utterance processed in') - BRAIN.isMuted = false - const processedData = await NLU.process(utterance) + const processedData = await this.getSocketNlu(socket.id)?.process( + utterance + ) if (processedData) { Telemetry.utterance(processedData) @@ -130,9 +143,9 @@ export default class SocketServer { }) // Handle automatic speech recognition - this.socket?.on('recognize', async (data: Buffer) => { + socket.on('recognize', async (data: Buffer) => { try { - await ASR.encode(data) + await ASR.encode(data, socket) } catch (e) { LogHelper.error( `ASR - Failed to encode audio blob to WAVE file: ${e}` @@ -142,9 +155,8 @@ export default class SocketServer { } }) - this.socket.once('disconnect', () => { - // TODO - // deleteProvider(this.socket.id) + socket.once('disconnect', () => { + this.onDisconnect(socket) }) }) } diff --git a/server/src/core/stt/stt.ts b/server/src/core/stt/stt.ts index 981f3cc354..5516fe020b 100644 --- a/server/src/core/stt/stt.ts +++ b/server/src/core/stt/stt.ts @@ -4,10 +4,12 @@ import path from 'node:path' import type { ASRAudioFormat } from '@/core/asr/types' import type { STTParser } from '@/core/stt/types' import { STT_PROVIDER, VOICE_CONFIG_PATH } from '@/constants' -import { SOCKET_SERVER, ASR } from '@/core' +import { ASR } from '@/core' import { STTParserNames, STTProviders } from '@/core/stt/types' import { LogHelper } from '@/helpers/log-helper' +import { ClientSocket } from '../socket-server' + const PROVIDERS_MAP = { [STTProviders.GoogleCloudSTT]: STTParserNames.GoogleCloudSTT, [STTProviders.WatsonSTT]: STTParserNames.WatsonSTT, @@ -85,7 +87,10 @@ export default class STT { /** * Read the speech file and transcribe */ - public async transcribe(audioFilePath: string): Promise { + public async transcribe( + audioFilePath: string, + socket: ClientSocket + ): Promise { LogHelper.info('Parsing WAVE file...') if (!fs.existsSync(audioFilePath)) { @@ -99,7 +104,7 @@ export default class STT { if (transcript && transcript !== '') { // Forward the string to the client - this.forward(transcript) + this.forward(transcript, socket) } else { this.deleteAudios() } @@ -111,8 +116,8 @@ export default class STT { * Forward string output to the client * and delete audio files once it has been forwarded */ - private forward(str: string): void { - SOCKET_SERVER.socket?.emit('recognized', str, (confirmation: string) => { + private forward(str: string, socket: ClientSocket): void { + socket.emit('recognized', str, (confirmation: string) => { if (confirmation === 'string-received') { this.deleteAudios() } diff --git a/server/src/core/tts/synthesizers/amazon-polly-synthesizer.ts b/server/src/core/tts/synthesizers/amazon-polly-synthesizer.ts index af45427a12..7e953d932d 100644 --- a/server/src/core/tts/synthesizers/amazon-polly-synthesizer.ts +++ b/server/src/core/tts/synthesizers/amazon-polly-synthesizer.ts @@ -8,7 +8,7 @@ import type { LongLanguageCode } from '@/types' import type { SynthesizeResult } from '@/core/tts/types' import type { AmazonVoiceConfigurationSchema } from '@/schemas/voice-config-schemas' import { LANG, VOICE_CONFIG_PATH, TMP_PATH } from '@/constants' -import { TTS } from '@/core' +import TextToSpeech from '@/core/tts/tts' import { TTSSynthesizerBase } from '@/core/tts/tts-synthesizer-base' import { LogHelper } from '@/helpers/log-helper' import { StringHelper } from '@/helpers/string-helper' @@ -27,7 +27,11 @@ export default class AmazonPollySynthesizer extends TTSSynthesizerBase { protected readonly lang = LANG as LongLanguageCode private readonly client: Polly | undefined = undefined - constructor(lang: LongLanguageCode) { + private _tts: TextToSpeech + public get tts(): TextToSpeech { + return this._tts + } + constructor(tts: TextToSpeech, lang: LongLanguageCode) { super() LogHelper.title(this.name) @@ -37,6 +41,8 @@ export default class AmazonPollySynthesizer extends TTSSynthesizerBase { fs.readFileSync(path.join(VOICE_CONFIG_PATH, 'amazon.json'), 'utf8') ) + this._tts = tts + try { this.lang = lang this.client = new Polly(config) @@ -81,7 +87,7 @@ export default class AmazonPollySynthesizer extends TTSSynthesizerBase { const duration = await this.getAudioDuration(audioFilePath) - TTS.em.emit('saved', duration) + this.tts.em.emit('saved', duration) return { audioFilePath, diff --git a/server/src/core/tts/synthesizers/flite-synthesizer.ts b/server/src/core/tts/synthesizers/flite-synthesizer.ts index 93e387a4e4..4088723e51 100644 --- a/server/src/core/tts/synthesizers/flite-synthesizer.ts +++ b/server/src/core/tts/synthesizers/flite-synthesizer.ts @@ -5,7 +5,7 @@ import { spawn } from 'node:child_process' import type { LongLanguageCode } from '@/types' import type { SynthesizeResult } from '@/core/tts/types' import { LANG, TMP_PATH, BIN_PATH } from '@/constants' -import { TTS } from '@/core' +import TextToSpeech from '@/core/tts/tts' import { TTSSynthesizerBase } from '@/core/tts/tts-synthesizer-base' import { LogHelper } from '@/helpers/log-helper' import { StringHelper } from '@/helpers/string-helper' @@ -22,7 +22,11 @@ export default class FliteSynthesizer extends TTSSynthesizerBase { protected readonly lang = LANG as LongLanguageCode private readonly binPath = path.join(BIN_PATH, 'flite', 'flite') - constructor(lang: LongLanguageCode) { + private _tts: TextToSpeech + public get tts(): TextToSpeech { + return this._tts + } + constructor(tts: TextToSpeech, lang: LongLanguageCode) { super() LogHelper.title(this.name) @@ -36,6 +40,8 @@ export default class FliteSynthesizer extends TTSSynthesizerBase { ) } + this._tts = tts + if (!fs.existsSync(this.binPath)) { LogHelper.error( `Cannot find ${this.binPath} You can set up the offline TTS by running: "npm run setup:offline-tts"` @@ -71,7 +77,7 @@ export default class FliteSynthesizer extends TTSSynthesizerBase { const duration = await this.getAudioDuration(audioFilePath) - TTS.em.emit('saved', duration) + this.tts.em.emit('saved', duration) return { audioFilePath, diff --git a/server/src/core/tts/synthesizers/google-cloud-tts-synthesizer.ts b/server/src/core/tts/synthesizers/google-cloud-tts-synthesizer.ts index bada947198..6d8693f2ca 100644 --- a/server/src/core/tts/synthesizers/google-cloud-tts-synthesizer.ts +++ b/server/src/core/tts/synthesizers/google-cloud-tts-synthesizer.ts @@ -2,13 +2,13 @@ import path from 'node:path' import fs from 'node:fs' import type { TextToSpeechClient } from '@google-cloud/text-to-speech' -import tts from '@google-cloud/text-to-speech' +import gTTS from '@google-cloud/text-to-speech' import { google } from '@google-cloud/text-to-speech/build/protos/protos' import type { LongLanguageCode } from '@/types' +import TextToSpeech from '@/core/tts/tts' import type { SynthesizeResult } from '@/core/tts/types' import { LANG, VOICE_CONFIG_PATH, TMP_PATH } from '@/constants' -import { TTS } from '@/core' import { TTSSynthesizerBase } from '@/core/tts/tts-synthesizer-base' import { LogHelper } from '@/helpers/log-helper' import { StringHelper } from '@/helpers/string-helper' @@ -34,7 +34,11 @@ export default class GoogleCloudTTSSynthesizer extends TTSSynthesizerBase { protected readonly lang = LANG as LongLanguageCode private readonly client: TextToSpeechClient | undefined = undefined - constructor(lang: LongLanguageCode) { + private _tts: TextToSpeech + public get tts(): TextToSpeech { + return this._tts + } + constructor(tts: TextToSpeech, lang: LongLanguageCode) { super() LogHelper.title(this.name) @@ -45,9 +49,11 @@ export default class GoogleCloudTTSSynthesizer extends TTSSynthesizerBase { 'google-cloud.json' ) + this._tts = tts + try { this.lang = lang - this.client = new tts.TextToSpeechClient() + this.client = new gTTS.TextToSpeechClient() LogHelper.success('Synthesizer initialized') } catch (e) { @@ -81,7 +87,7 @@ export default class GoogleCloudTTSSynthesizer extends TTSSynthesizerBase { const duration = await this.getAudioDuration(audioFilePath) - TTS.em.emit('saved', duration) + this.tts.em.emit('saved', duration) return { audioFilePath, diff --git a/server/src/core/tts/synthesizers/watson-tts-synthesizer.ts b/server/src/core/tts/synthesizers/watson-tts-synthesizer.ts index 3990da8a9a..42e600dd8c 100644 --- a/server/src/core/tts/synthesizers/watson-tts-synthesizer.ts +++ b/server/src/core/tts/synthesizers/watson-tts-synthesizer.ts @@ -8,7 +8,7 @@ import type { WatsonVoiceConfigurationSchema } from '@/schemas/voice-config-sche import type { LongLanguageCode } from '@/types' import type { SynthesizeResult } from '@/core/tts/types' import { LANG, VOICE_CONFIG_PATH, TMP_PATH } from '@/constants' -import { TTS } from '@/core' +import TextToSpeech from '@/core/tts/tts' import { TTSSynthesizerBase } from '@/core/tts/tts-synthesizer-base' import { LogHelper } from '@/helpers/log-helper' import { StringHelper } from '@/helpers/string-helper' @@ -27,7 +27,11 @@ export default class WatsonTTSSynthesizer extends TTSSynthesizerBase { protected readonly lang: LongLanguageCode = LANG as LongLanguageCode private readonly client: Tts | undefined = undefined - constructor(lang: LongLanguageCode) { + private _tts: TextToSpeech + public get tts(): TextToSpeech { + return this._tts + } + constructor(tts: TextToSpeech, lang: LongLanguageCode) { super() LogHelper.title(this.name) @@ -37,6 +41,7 @@ export default class WatsonTTSSynthesizer extends TTSSynthesizerBase { fs.readFileSync(path.join(VOICE_CONFIG_PATH, 'watson-stt.json'), 'utf8') ) + this._tts = tts try { this.lang = lang this.client = new Tts({ @@ -75,7 +80,7 @@ export default class WatsonTTSSynthesizer extends TTSSynthesizerBase { const duration = await this.getAudioDuration(audioFilePath) - TTS.em.emit('saved', duration) + this.tts.em.emit('saved', duration) return { audioFilePath, diff --git a/server/src/core/tts/tts.ts b/server/src/core/tts/tts.ts index d120e0ae2e..8282d9af02 100644 --- a/server/src/core/tts/tts.ts +++ b/server/src/core/tts/tts.ts @@ -4,7 +4,7 @@ import fs from 'node:fs' import type { ShortLanguageCode } from '@/types' import type { TTSSynthesizer } from '@/core/tts/types' -import { SOCKET_SERVER } from '@/core' +import Brain from '@/core/brain/brain' import { TTS_PROVIDER, VOICE_CONFIG_PATH } from '@/constants' import { TTSSynthesizers, TTSProviders } from '@/core/tts/types' import { LogHelper } from '@/helpers/log-helper' @@ -31,13 +31,18 @@ export default class TTS { public lang: ShortLanguageCode = 'en' public em = new events.EventEmitter() - constructor() { + private _brain: Brain + public get brain(): Brain { + return this._brain + } + constructor(brain: Brain) { if (!TTS.instance) { LogHelper.title('TTS') LogHelper.success('New instance') TTS.instance = this } + this._brain = brain } /** @@ -112,7 +117,7 @@ export default class TTS { const { audioFilePath, duration } = result const bitmap = await fs.promises.readFile(audioFilePath) - SOCKET_SERVER.socket?.emit( + this.brain.socket?.emit( 'audio-forwarded', { buffer: Buffer.from(bitmap),