start of attempt at AWS Transcribe Streaming

index.js

@@ -46,7 +46,7 @@ server.on("connection", (client) => {
 		codecs: codecs,
 		languages: languages,
 		transport: client,
-		provider: getProvider("google", argv),
+		provider: getProvider("aws", argv),
 	});
 });
 

lib/provider.js

@@ -15,8 +15,17 @@
  *  limitations under the License.
  */
 
-const { Writable } = require('stream');
-const speech = require('@google-cloud/speech');
+const {
+	Writable,
+	Transform,
+	PassThrough,
+} = require('stream');
+const GoogleSpeech = require('@google-cloud/speech');
+const {
+	TranscribeStreamingClient,
+	StartStreamTranscriptionCommand,
+} = require('@aws-sdk/client-transcribe-streaming');
+const { config } = require('process');
 
 /*
  * For speech provider implementer.
@@ -114,7 +123,7 @@ class GoogleProvider extends Writable {
 	}
 
 	_construct(callback) {
-		this.client = new speech.SpeechClient();
+		this.client = new GoogleSpeech.SpeechClient();
 
 		callback();
 	}
@@ -270,6 +279,249 @@ class GoogleProvider extends Writable {
 		this.stop();
 		this.start(config);
 	}
+};
+
+// class AWSProvider extends Writable {
+// 	constructor(options) {
+// 		super();
+
+// 		this.LanguageCode = "en-GB";
+// 		this.MediaEncoding = "pcm";
+// 		this.credentials = {
+			// accessKeyId: process.env.AWS_ACCESS_KEY_ID,
+			// secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY,
+// 		}
+
+// 		this.results = [];
+// 		this.stream = new TransformStream();
+// 		this.recognizeStream = this.stream.writable.getWriter();
+
+// 		(async () => {
+// 			this.readStream = this.stream.readable;
+
+// 			this.config = {
+// 				LanguageCode: this.LanguageCode,
+// 				MediaEncoding: this.MediaEncoding,
+// 				MediaSampleRateHertz: "8000",
+// 				AudioStream: (async function* () {
+// 					for await (const chunk of this.stream.readable.read()) {
+// 						yield {AudioEvent: {AudioChunk: chunk}};
+// 					}
+// 					// for (let result = await this.readStream.read(); !result.done; result = await this.readStream.read()) {
+// 					// 	return {AudioEvent: {AudioChunk: result}};
+// 					// }
+// 				})(),
+// 				// AudioStream: this.readStream.read(),
+// 			}	
+// 		})
+
+
+// 		this.command = null;
+// 	}
+
+// 	_construct(callback) {
+// 		this.client = new TranscribeStreamingClient({
+// 			region: "eu-west-2",
+// 			credentials: this.credentials
+// 		});
+
+// 		callback();
+// 	}
+
+// 	_write(chunk, encoding, callback) {
+// 		if (this.recognizeStream) {
+// 			console.debug("writing chunk");
+// 			this.recognizeStream.write(chunk);
+// 		}
+
+// 		callback();
+// 	}
+
+// 	setConfig(config) {
+// 		if (!config) {
+// 			return;
+// 		}
+
+// 	}
+
+// 	start(config) {
+// 		if (this.command) {
+// 			return;
+// 		}
+
+// 		this.setConfig(config);
+// 		config = this.config;
+
+// 		// this.config.AudioStream = this.recognizeStream;
+
+// 		this.command = new StartStreamTranscriptionCommand(this.config)
+// 		console.debug("command created")
+
+// 		this.client.send(this.command).then((response) => {
+// 			print("response from command");
+// 			console.debug(response)
+// 		}).catch((err) => {
+// 			console.log("error in command");
+// 			console.debug(err);
+// 		})
+// 		// (async function() {
+// 		// 	// console.debug("start async")
+// 		// 	this.results = await this.client.send(this.command);
+// 		// 	// .then(async (response) => {
+// 		// 	// 	console.debug(JSON.stringify(response));
+// 		// 	// }).catch(err => {
+// 		// 	// 	console.debug("error");
+// 		// 	// 	console.debug(err);
+	
+// 		// 	// });
+// 		// 	try {
+// 		// 		// console.debug("event");
+// 		// 		for await (const event of this.results.TranscriptResultStream) {
+// 		// 			console.log(JSON.stringify(event));
+// 		// 		}
+// 		// 	} catch (err) {
+// 		// 		console.log("error")
+// 		// 		console.log(err)
+// 		// 	}
+// 		// })();
+
+// 		// console.debug("AWSProvider");
+// 	}
+
+// 	restart(config) {
+// 		this.start(config);
+// 	}
+// }
+
+class AWSProvider extends Writable {
+	constructor(options) {
+		super();
+
+		this.LanguageCode = "en-GB";
+		this.MediaEncoding = "pcm";
+		this.credentials = {
+			accessKeyId: process.env.AWS_ACCESS_KEY_ID,
+			secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY,
+		}
+
+		console.debug(process.env.AWS_SECRET_KEY_ID)
+
+		this.stream = new TransformStream();
+		this.readStream = this.stream.readable.getReader();
+		this.writeStream = this.stream.writable;
+
+		this.recognizeStream = null;
+
+
+		console.log(this.writeStream);
+	}
+
+	_construct(callback) {
+		this.client = new TranscribeStreamingClient({
+			region: "eu-west-2",
+			credentials: this.credentials
+		});
+
+		callback();
+	}
+
+	_write(chunk, encoding, callback) {
+		this.recognizeStream.write(chunk);
+
+		// this.readStream.read().then((res) => {
+		// 	console.log(res)
+		// })
+
+		callback();
+	}
+
+	_writev(chunks, callback) {
+		for (let chunk in chunks) {
+			this._write(chunk, null, callback);
+		}
+
+		callback();
+	}
+
+	_final(callback) {
+		this.stop();
+
+		callback();
+	}
+
+	start(config) {
+		// this.setConfig(config);
+		// config = this.config;
+
+		console.log("START");
+
+		this.recognizeStream = this.writeStream.getWriter();
+
+		// const passthrough = new PassThrough();
+		// this.readStream.pipe(passthrough);
+
+		const readStream = this.readStream;
+
+		async function* audioSource() {
+			// await readStream.start();
+			while (readStream.ends !== true) {
+				const chunk = await readStream.read();
+				yield chunk;
+			}
+		}
+
+		async function* audioStream() {
+			for await (const chunk of audioSource()) {
+				console.debug("CHUNKING");
+				yield {AudioEvent: {AudioChunk: chunk.value}};
+			}
+		}
+
+		audioStream().next().then(res => console.debug(res.value.AudioEvent.AudioChunk));
+
+		// console.debug('AUDIO');
+		// this.audioStream().next().then(res => console.debug(res));
+		// this.audioStream().next().then(res => console.debug(res));
+
+		this.param = {
+			LanguageCode: this.LanguageCode,
+			MediaEncoding: this.MediaEncoding,
+			MediaSampleRateHertz: 8000,
+			AudioStream: audioStream(),
+		}
+
+		const command = new StartStreamTranscriptionCommand(this.param);
+
+		this.client.send(command).then(async (res) => {
+			for await (const event of res.TranscriptResultStream) {
+				console.debug(event);
+			};
+		})
+		return;
+
+		(async () => {
+			console.log("STARTED");
+		})
+
+	}
+
+	stop() {
+		if(!this.recognizeStream) {
+			return;
+		}
+
+		// this.recognizeStream.close();
+		// console.log(this.recognizeStream);
+
+		console.log("End of stream");
+
+		// return;
+	}
+
+	restart(config) {
+		this.stop()
+		this.start(config)
+	}
 }
 
 /**
@@ -284,6 +536,10 @@ function getProvider(name, options) {
 		return new GoogleProvider(options);
 	}
 
+	if (name == "aws") {
+		return new AWSProvider(options);
+	}
+
 	throw new Error("Unsupported speech provider '" + name + "'");
 }
 

package.json

@@ -18,6 +18,7 @@
 		"speech"
 	],
 	"dependencies": {
+		"@aws-sdk/client-transcribe-streaming": "^3.362.0",
 		"@google-cloud/speech": "^4.9.0",
 		"ws": "^8.3.0",
 		"yargs": "^17.3.1"