Feat/assemblyai testing (#516)

* handle errors from assemblyai

* wip

* fix alert

* normalizeAssemblyai

lib/tasks/gather.js

@@ -466,6 +466,7 @@ class TaskGather extends SttTask {
         ep.addCustomEventListener(AssemblyAiTranscriptionEvents.Transcription,
           this._onTranscription.bind(this, cs, ep));
         ep.addCustomEventListener(AssemblyAiTranscriptionEvents.Connect, this._onVendorConnect.bind(this, cs, ep));
+        ep.addCustomEventListener(AssemblyAiTranscriptionEvents.Error, this._onVendorError.bind(this, cs, ep));
         ep.addCustomEventListener(AssemblyAiTranscriptionEvents.ConnectFailure,
           this._onVendorConnectFailure.bind(this, cs, ep));
         break;
@@ -797,10 +798,6 @@ class TaskGather extends SttTask {
   _onTranscriptionComplete(cs, ep) {
     this.logger.debug('TaskGather:_onTranscriptionComplete');
   }
-  _onVendorConnect(_cs, _ep) {
-    this.logger.debug(`TaskGather:_on${this.vendor}Connect`);
-  }
-
   async _onJambonzError(cs, ep, evt) {
     this.logger.info({evt}, 'TaskGather:_onJambonzError');
     if (this.isHandledByPrimaryProvider && this.fallbackVendor) {
@@ -835,19 +832,15 @@ class TaskGather extends SttTask {
   }
 
   _onVendorConnectFailure(cs, _ep, evt) {
-    const {reason} = evt;
+    super._onVendorConnectFailure(cs, _ep, evt);
-    const {writeAlerts, AlertType} = cs.srf.locals;
-    this.logger.info({evt}, `TaskGather:_on${this.vendor}ConnectFailure`);
-    writeAlerts({
-      account_sid: cs.accountSid,
-      alert_type: AlertType.STT_FAILURE,
-      message: `Failed connecting to ${this.vendor} speech recognizer: ${reason}`,
-      vendor: this.vendor,
-    }).catch((err) => this.logger.info({err}, `Error generating alert for ${this.vendor} connection failure`));
-    this.notifyError({msg: 'ASR error', details:`Failed connecting to speech vendor ${this.vendor}: ${reason}`});
     this.notifyTaskDone();
   }
 
+  _onVendorError(cs, _ep, evt) {
+    super._onVendorError(cs, _ep, evt);
+    this._resolve('stt-error', evt);
+  }
+
   _onVadDetected(cs, ep) {
     if (this.bargein && this.minBargeinWordCount === 0) {
       this.logger.debug('TaskGather:_onVadDetected');
@@ -921,6 +914,13 @@ class TaskGather extends SttTask {
           await this.performAction({reason: 'timeout'});
         }
       }
+      else if (reason.startsWith('stt-error')) {
+        if (this.parentTask) this.parentTask.emit('stt-error', evt);
+        else {
+          this.emit('stt-error', evt);
+          await this.performAction({reason: 'error', details: evt.error});
+        }
+      }
     } catch (err) {  /*already logged error*/ }
     this.notifyTaskDone();
   }

lib/tasks/stt-task.js

@@ -148,6 +148,36 @@ class SttTask extends Task {
     const dgOptions = this.data.recognizer.deepgramOptions = this.data.recognizer.deepgramOptions || {};
     dgOptions.utteranceEndMs = dgOptions.utteranceEndMs || asrTimeout;
   }
+
+  _onVendorConnect(_cs, _ep) {
+    this.logger.debug(`TaskGather:_on${this.vendor}Connect`);
+  }
+
+  _onVendorError(cs, _ep, evt) {
+    this.logger.info({evt}, `${this.name}:_on${this.vendor}Error`);
+    const {writeAlerts, AlertType} = cs.srf.locals;
+    writeAlerts({
+      account_sid: cs.accountSid,
+      alert_type: AlertType.STT_FAILURE,
+      message: 'STT failure reported by vendor',
+      detail: evt.error,
+      vendor: this.vendor,
+    }).catch((err) => this.logger.info({err}, `Error generating alert for ${this.vendor} connection failure`));
+    this.notifyError({msg: 'ASR error', details:`Failed connecting to speech vendor ${this.vendor}: ${evt.error}`});
+  }
+
+  _onVendorConnectFailure(cs, _ep, evt) {
+    const {reason} = evt;
+    const {writeAlerts, AlertType} = cs.srf.locals;
+    this.logger.info({evt}, `${this.name}:_on${this.vendor}ConnectFailure`);
+    writeAlerts({
+      account_sid: cs.accountSid,
+      alert_type: AlertType.STT_FAILURE,
+      message: `Failed connecting to ${this.vendor} speech recognizer: ${reason}`,
+      vendor: this.vendor,
+    }).catch((err) => this.logger.info({err}, `Error generating alert for ${this.vendor} connection failure`));
+    this.notifyError({msg: 'ASR error', details:`Failed connecting to speech vendor ${this.vendor}: ${reason}`});
+  }
 }
 
 module.exports = SttTask;

lib/tasks/transcribe.js

@@ -298,8 +298,8 @@ class TaskTranscribe extends SttTask {
         this.bugname = 'assemblyai_transcribe';
         ep.addCustomEventListener(AssemblyAiTranscriptionEvents.Transcription,
           this._onTranscription.bind(this, cs, ep, channel));
-        ep.addCustomEventListener(AssemblyAiTranscriptionEvents.Connect,
+        ep.addCustomEventListener(AssemblyAiTranscriptionEvents.Connect, this._onVendorConnect.bind(this, cs, ep));
-          this._onVendorConnect.bind(this, cs, ep));
+        ep.addCustomEventListener(AssemblyAiTranscriptionEvents.Error, this._onVendorError.bind(this, cs, ep));
         ep.addCustomEventListener(AssemblyAiTranscriptionEvents.ConnectFailure,
           this._onVendorConnectFailure.bind(this, cs, ep, channel));
         break;
@@ -529,22 +529,8 @@ class TaskTranscribe extends SttTask {
     }
   }
 
-  _onVendorConnect(_cs, _ep) {
-    this.logger.debug(`TaskTranscribe:_on${this.vendor}Connect`);
-  }
-
   _onVendorConnectFailure(cs, _ep, channel, evt) {
-    const {reason} = evt;
+    super._onVendorConnectFailure(cs, _ep, evt);
-    const {writeAlerts, AlertType} = cs.srf.locals;
-    this.logger.info({evt}, `TaskTranscribe:_on${this.vendor}ConnectFailure`);
-    writeAlerts({
-      account_sid: cs.accountSid,
-      alert_type: AlertType.STT_FAILURE,
-      message: `Failed connecting to ${this.vendor} speech recognizer: ${reason}`,
-      vendor: this.vendor,
-    }).catch((err) => this.logger.info({err}, `Error generating alert for ${this.vendor} connection failure`));
-    this.notifyError(`Failed connecting to speech vendor ${this.vendor}: ${reason}`);
-
     if (this.childSpan[channel - 1] && this.childSpan[channel - 1].span) {
       this.childSpan[channel - 1].span.setAttributes({
         channel,
@@ -555,10 +541,6 @@ class TaskTranscribe extends SttTask {
     this.notifyTaskDone();
   }
 
-  _onVendorError(cs, _ep, _channel, vendor, evt) {
-    this.logger.info({evt}, `TaskTranscribe:_on${vendor}Error`);
-  }
-
   _startAsrTimer(channel) {
     if (this.vendor === 'deepgram') return; // no need
     assert(this.isContinuousAsr);

lib/utils/constants.json

@@ -128,6 +128,7 @@
   },
   "AssemblyAiTranscriptionEvents": {
     "Transcription": "assemblyai_transcribe::transcription",
+    "Error": "assemblyai_transcribe::error",
     "ConnectFailure": "assemblyai_transcribe::connect_failed",
     "Connect": "assemblyai_transcribe::connect"
   },

lib/utils/transcription-utils.js

@@ -380,6 +380,24 @@ const normalizeAws = (evt, channel, language) => {
   };
 };
 
+const normalizeAssemblyAi = (evt, channel, language) => {
+  const copy = JSON.parse(JSON.stringify(evt));
+  return {
+    language_code: language,
+    channel_tag: channel,
+    is_final: evt.message_type === 'FinalTranscript',
+    alternatives: [
+      {
+        confidence: evt.confidence,
+        transcript: evt.text,
+      }
+    ],
+    vendor: {
+      name: 'ASSEMBLYAI',
+      evt: copy
+    }
+  };
+};
 
 module.exports = (logger) => {
   const normalizeTranscription = (evt, vendor, channel, language, shortUtterance) => {
@@ -404,6 +422,8 @@ module.exports = (logger) => {
         return normalizeSoniox(evt, channel, language);
       case 'cobalt':
         return normalizeCobalt(evt, channel, language);
+      case 'assemblyai':
+        return normalizeAssemblyAi(evt, channel, language, shortUtterance);
       default:
         if (vendor.startsWith('custom:')) {
           return normalizeCustom(evt, channel, language, vendor);