Merge pull request #26 from codeforequity-at/bugfix/BOT-3625-voice-stt-throttling---error-ha

murliwatz · web-flow · commit fab3a709e0e9 · 2023-03-10T10:20:12.000+01:00
Bugfix/bot 3625 voice stt throttling   error ha
diff --git a/docker-compose-frontend.yml b/docker-compose-frontend.yml
@@ -0,0 +1,18 @@
+version: '3'
+services:
+  frontend:
+    image: botium/botium-speech-frontend:develop
+    restart: always
+    user: 1000:1000
+    ports:
+    - 80:56000
+    environment:
+      BOTIUM_API_TOKENS:
+      BOTIUM_SPEECH_PROVIDER_TTS: marytts
+      BOTIUM_SPEECH_PROVIDER_STT: kaldi
+      BOTIUM_SPEECH_GOOGLE_CLIENT_EMAIL:
+      BOTIUM_SPEECH_GOOGLE_PRIVATE_KEY:
+    volumes:
+      - "./frontend/resources:/app/resources"
+volumes:
+  logs:
diff --git a/frontend/src/routes.js b/frontend/src/routes.js
@@ -885,7 +885,8 @@ const wssUpgrade = (req, socket, head) => {
 
   const wss1 = new WebSocketServer({ noServer: true })
   wss1.on('connection', async (ws) => {
-    stream.events.on('data', (data) => {
+    stream.events.on('data', async (data) => {
+      if (data.err) debug(data)
       data.streamDuration = ((new Date() - stream.dateTimeStart) / 1000).toFixed(3)
       ws.send(JSON.stringify(data))
     })
@@ -903,6 +904,7 @@ const wssUpgrade = (req, socket, head) => {
       stream.close()
       wss1.close()
     })
+    stream.triggerHistoryEmit()
   })
   wss1.handleUpgrade(req, socket, head, (ws) => {
     wss1.emit('connection', ws, req)
diff --git a/frontend/src/stt/awstranscribe.js b/frontend/src/stt/awstranscribe.js
@@ -78,6 +78,7 @@ class AwsTranscribeSTT {
     applyIfExists(request, req, 'req.body.awstranscribe.config.streaming')
 
     const events = new EventEmitter()
+    let eventHistory = []
     try {
       const cmdResponse = await transcribeClient.send(new StartStreamTranscriptionCommand(request))
       setTimeout(async () => {
@@ -88,21 +89,26 @@ class AwsTranscribeSTT {
             if (results && results.length > 0) {
               for (const result of results) {
                 const event = {
+                  status: 'ok',
                   text: result.Alternatives[0].Transcript,
                   final: !result.IsPartial,
                   start: result.StartTime,
                   end: result.EndTime,
                   debug: result
                 }
                 events.emit('data', event)
+                eventHistory.push(event)
               }
             }
           }
         } catch (err) {
           debug(`TranscriptResultStream failure: ${err.Message || err.message || err}`)
-          events.emit('data', {
-            err: `${err.message || err}`
-          })
+          const event = {
+            status: 'error',
+            err: `AWS Transcribe STT failed: ${err.message || err}`
+          }
+          events.emit('data', event)
+          eventHistory.push(event)
         }
         events.emit('close')
         debug('Ready listening to TranscriptResultStream ')
@@ -126,6 +132,12 @@ class AwsTranscribeSTT {
           audioInputStream.destroy()
         }
         audioInputStream = null
+        eventHistory = null
+      },
+      triggerHistoryEmit: () => {
+        for (const eh of eventHistory) {
+          events.emit('data', eh)
+        }
       }
     }
   }
diff --git a/frontend/src/stt/azure.js b/frontend/src/stt/azure.js
@@ -52,46 +52,56 @@ class AzureSTT {
     const recognizer = new SpeechRecognizer(speechConfig, audioConfig)
 
     const events = new EventEmitter()
+    let eventHistory = []
 
     const recognizedHandler = (s, e) => {
       if (e.result.reason === ResultReason.RecognizedSpeech || e.result.reason === ResultReason.RecognizingSpeech) {
         const event = {
+          status: 'ok',
           text: e.result.text,
           final: e.result.reason === ResultReason.RecognizedSpeech,
           debug: e.result
         }
         event.start = _.round(e.result.offset / 10000000, 3)
         event.end = _.round((e.result.offset + e.result.duration) / 10000000, 3)
         events.emit('data', event)
+        eventHistory.push(event)
       }
     }
     recognizer.recognizing = recognizedHandler
     recognizer.recognized = recognizedHandler
     recognizer.sessionStopped = (s, e) => {
       recognizer.stopContinuousRecognitionAsync()
-      events.emit('close')
+    }
+    recognizer.canceled = (s, e) => {
+      const event = {
+        status: 'error',
+        err: `Azure STT failed: ${getAzureErrorDetails(e)}`
+      }
+      events.emit('data', event)
+      eventHistory.push(event)
     }
     recognizer.startContinuousRecognitionAsync()
 
     return new Promise((resolve, reject) => {
-      recognizer.canceled = (s, e) => {
-        recognizer.stopContinuousRecognitionAsync()
-        reject(new Error(`Azure STT failed: ${getAzureErrorDetails(e)}`))
-      }
-      recognizer.sessionStarted = (s, e) => {
-        resolve({
-          events,
-          write: (buffer) => {
-            pushStream.write(buffer)
-          },
-          end: () => {
-          },
-          close: () => {
-            recognizer.stopContinuousRecognitionAsync()
-            pushStream.close()
+      resolve({
+        events,
+        write: (buffer) => {
+          pushStream.write(buffer)
+        },
+        end: () => {
+        },
+        close: () => {
+          recognizer.stopContinuousRecognitionAsync()
+          pushStream.close()
+          eventHistory = null
+        },
+        triggerHistoryEmit: () => {
+          for (const eh of eventHistory) {
+            events.emit('data', eh)
           }
-        })
-      }
+        }
+      })
     })
   }
 
diff --git a/frontend/src/stt/google.js b/frontend/src/stt/google.js
@@ -73,11 +73,13 @@ class GoogleSTT {
     const bufferStream = new PassThrough()
     bufferStream.pipe(recognizeStream)
     const events = new EventEmitter()
+    let eventHistory = []
 
     recognizeStream.on('data', (data) => {
       const alternative = data.results[0] && data.results[0].alternatives[0]
       if (alternative && alternative.transcript) {
         const event = {
+          status: 'ok',
           text: alternative.transcript,
           final: !!data.results[0].isFinal,
           debug: data
@@ -87,12 +89,16 @@ class GoogleSTT {
           event.end = _.round(_.toNumber(alternative.words[alternative.words.length - 1].endTime.seconds) + _.toNumber(alternative.words[alternative.words.length - 1].endTime.nanos) / 1000000000, 3)
         }
         events.emit('data', event)
+        eventHistory.push(event)
       }
     })
     recognizeStream.on('error', (err) => {
-      events.emit('data', {
-        err: `${err.message}`
-      })
+      const event = {
+        status: 'error',
+        err: `Google STT failed: ${err.message}`
+      }
+      events.emit('data', event)
+      eventHistory.push(event)
     })
     recognizeStream.on('close', () => {
       events.emit('close')
@@ -112,7 +118,13 @@ class GoogleSTT {
         if (recognizeStream) {
           recognizeStream.destroy()
         }
+        eventHistory = null
         recognizeStream = null
+      },
+      triggerHistoryEmit: () => {
+        for (const eh of eventHistory) {
+          events.emit('data', eh)
+        }
       }
     }
   }
diff --git a/frontend/src/stt/ibm.js b/frontend/src/stt/ibm.js
@@ -42,12 +42,14 @@ class IbmSTT {
     const bufferStream = new PassThrough()
     bufferStream.pipe(recognizeStream)
     const events = new EventEmitter()
+    let eventHistory = []
 
     recognizeStream.on('data', (data) => {
       for (const result of data.results || []) {
         const transcription = result.alternatives[0] ? result.alternatives[0].transcript : null
         if (transcription) {
           const event = {
+            status: 'ok',
             text: transcription,
             final: !!result.final,
             debug: result
@@ -57,13 +59,17 @@ class IbmSTT {
             event.end = _.round(result.alternatives[result.alternatives.length - 1].timestamps[0][2], 3)
           }
           events.emit('data', event)
+          eventHistory.push(event)
         }
       }
     })
     recognizeStream.on('error', (err) => {
-      events.emit('data', {
-        err: `${err.message}`
-      })
+      const event = {
+        status: 'error',
+        err: `IBM STT failed: ${err.message}`
+      }
+      events.emit('data', event)
+      eventHistory.push(event)
     })
     recognizeStream.on('close', () => {
       events.emit('close')
@@ -85,6 +91,12 @@ class IbmSTT {
           recognizeStream.destroy()
         }
         recognizeStream = null
+        eventHistory = null
+      },
+      triggerHistoryEmit: () => {
+        for (const eh of eventHistory) {
+          events.emit('data', eh)
+        }
       }
     }
   }
diff --git a/frontend/src/stt/kaldi.js b/frontend/src/stt/kaldi.js
@@ -42,6 +42,7 @@ class KaldiSTT {
 
       const ws = new WebSocket(wsUri)
       const events = new EventEmitter()
+      let eventHistory = []
 
       ws.on('open', () => {
         ws.on('message', (data) => {
@@ -50,6 +51,7 @@ class KaldiSTT {
             const hypotheses = dj.result && dj.result.hypotheses && dj.result.hypotheses[0]
             if (hypotheses && hypotheses.transcript) {
               const event = {
+                status: 'ok',
                 text: hypotheses.transcript,
                 final: !!dj.result.final,
                 debug: dj
@@ -59,15 +61,19 @@ class KaldiSTT {
                 event.end = _.round(_.toNumber(dj['total-length']), 3)
               }
               events.emit('data', event)
+              eventHistory.push(event)
             }
           } catch (err) {
             debug(`received non JSON content on stream, ignoring. ${err.message}`)
           }
         })
         ws.on('error', (err) => {
-          events.emit('data', {
-            err: `${err.message}`
-          })
+          const event = {
+            status: 'error',
+            err: `Kaldi STT failed: ${err.message}`
+          }
+          events.emit('data', event)
+          eventHistory.push(event)
         })
         ws.on('close', () => {
           events.emit('close')
@@ -89,6 +95,12 @@ class KaldiSTT {
             if (ws) {
               ws.close()
             }
+            eventHistory = null
+          },
+          triggerHistoryEmit: () => {
+            for (const eh of eventHistory) {
+              events.emit('data', eh)
+            }
           }
         })
       })

Original file line number	Diff line number	Diff line change
`@@ -78,6 +78,7 @@ class AwsTranscribeSTT {`
`78`	`78`	`applyIfExists(request, req, 'req.body.awstranscribe.config.streaming')`
`79`	`79`
`80`	`80`	`const events = new EventEmitter()`
	`81`	`+ let eventHistory = []`
`81`	`82`	`try {`
`82`	`83`	`const cmdResponse = await transcribeClient.send(new StartStreamTranscriptionCommand(request))`
`83`	`84`	`setTimeout(async () => {`
`@@ -88,21 +89,26 @@ class AwsTranscribeSTT {`
`88`	`89`	`if (results && results.length > 0) {`
`89`	`90`	`for (const result of results) {`
`90`	`91`	`const event = {`
	`92`	`+ status: 'ok',`
`91`	`93`	`text: result.Alternatives[0].Transcript,`
`92`	`94`	`final: !result.IsPartial,`
`93`	`95`	`start: result.StartTime,`
`94`	`96`	`end: result.EndTime,`
`95`	`97`	`debug: result`
`96`	`98`	`}`
`97`	`99`	`events.emit('data', event)`
	`100`	`+ eventHistory.push(event)`
`98`	`101`	`}`
`99`	`102`	`}`
`100`	`103`	`}`
`101`	`104`	`} catch (err) {`
`102`	`105`	debug(`TranscriptResultStream failure: ${err.Message \|\| err.message \|\| err}`)
`103`		`- events.emit('data', {`
`104`		- err: `${err.message \|\| err}`
`105`		`- })`
	`106`	`+ const event = {`
	`107`	`+ status: 'error',`
	`108`	+ err: `AWS Transcribe STT failed: ${err.message \|\| err}`
	`109`	`+ }`
	`110`	`+ events.emit('data', event)`
	`111`	`+ eventHistory.push(event)`
`106`	`112`	`}`
`107`	`113`	`events.emit('close')`
`108`	`114`	`debug('Ready listening to TranscriptResultStream ')`
`@@ -126,6 +132,12 @@ class AwsTranscribeSTT {`
`126`	`132`	`audioInputStream.destroy()`
`127`	`133`	`}`
`128`	`134`	`audioInputStream = null`
	`135`	`+ eventHistory = null`
	`136`	`+ },`
	`137`	`+ triggerHistoryEmit: () => {`
	`138`	`+ for (const eh of eventHistory) {`
	`139`	`+ events.emit('data', eh)`
	`140`	`+ }`
`129`	`141`	`}`
`130`	`142`	`}`
`131`	`143`	`}`