asr/app/asr_engine.py

from traceback import print_exc
import base64
from appPublic.log import info, debug, warning, error, exception, critical
from appPublic.dictObject import DictObject
from appPublic.folderUtils import temp_file
from ahserver.serverenv import ServerEnv
from aiohttp.web import StreamResponse

from io import BytesIO
import struct

def audio_dic2list(audio):
	ks = [k for k in audio.keys()]
	ks.sort()
	return [audio[k] for k in ks]

def float32array_to_wav(samples, sample_rate=16000, num_channels=1):
	# Calculate the total number of samples
	num_samples = len(samples)

	# Calculate the byte rate
	byte_rate = sample_rate * num_channels * 4

	# Calculate the block align
	block_align = num_channels * 4

	# Create the WAV header
	header = struct.pack(
		'<4sI4s4sIHHIIHH4sI',
		b'RIFF', 36 + num_samples * 4, b'WAVE', b'fmt ', 16, 3, num_channels, sample_rate,
		byte_rate, block_align, 32, b'data', num_samples * 4
	)

	info(f'float32array_to_wav({samples[:10]}, ...)')
	# Convert the Float32Array to bytes
	data = struct.pack('f' * num_samples, *samples)

	# Write the header and data to a file
	tmpfile = temp_file(suffix='.wav')
	with open(tmpfile, 'wb') as f:
		f.write(header)
		f.write(data)
	return tmpfile

async def generate(request, **kw):
	params_kw = kw.get('params_kw', DictObject())
	info(f'{params_kw=}')
	model = params_kw.model
	audio = params_kw.audio
	if audio is None:
		return {
			'status':'error',
			'message':'audio is null'
		}
	engine = None
	g = ServerEnv()
	if model=='whisper':
		engine = g.whisper_engine

	if engine is None:
		return {
			'status':'error',
			'message':f'model={model} is not defined'
		}
	try:
		audio = audio_dic2list(audio)
		fname = float32array_to_wav(audio)
		txt = await engine.stt(fname)
		os.remove(fname)
		info(f'{audio=}, {txt=}')
		return {
			"status":"ok",
			"content":txt
		}
	except Exception as e:
		exception(f'{e}')
		print_exc()
		return {
			'status':'error',
			'message':f'{e}'
		}