import * as download from 'download' ; import * as temp from 'temp' ; import { Chat } from '@/models/Chat' ; import { Worker } from 'cluster' ; import { cpus } from 'os' ; import { v4 as uuid } from 'uuid' ; import { writeFileSync } from 'fs' ; import Cluster from '@/helpers/Cluster' ; import RecognitionResult from '@/helpers/engine/RecognitionResult' ; import RecognitionResultPart from '@/helpers/engine/RecognitionResultPart' ; import augmentError from '@/helpers/augmentError' ; import deleteFile from '@/helpers/deleteFile' ; import flac from '@/helpers/flac' ; import getTextFromAudio from '@/helpers/getTextFromAudio' ; const numCPUs = cpus ( ) . length ; interface WorkerMessage { url : string ; promiseId : string ; chat : Partial < Chat > ; userId ? : number ; forwardedFrom ? : string ; } interface PrimaryMessage { textWithTimecodes ? : RecognitionResultPart [ ] ; duration ? : number ; promiseId : string ; errorString ? : string ; } const workers = [ ] ; if ( Cluster . isPrimary ) { console . info ( `Primary ${ process . pid } is running` ) ; for ( let i = 0 ; i < numCPUs ; i += 1 ) { const worker = Cluster . fork ( ) ; worker . on ( 'message' , primaryReceivesMessage ) ; workers . push ( worker ) ; } } else { console . info ( `Worker ${ process . pid } started` ) ; process . on ( 'message' , workerReceivesMessage ) ; } const recognitionPromises = { } ; let clusterNumber = 0 ; export default function urlToText ( url , chat , userId ? , forwardedFrom ? ) { if ( clusterNumber >= workers . length ) { clusterNumber = 0 ; } const worker = workers [ clusterNumber ] ; clusterNumber += 1 ; return new Promise < RecognitionResult > ( ( res , rej ) => { const promiseId = uuid ( ) ; recognitionPromises [ promiseId ] = { res , rej } ; const workerMessage = { url , promiseId , chat , userId , forwardedFrom , } ; worker . send ( workerMessage ) ; } ) ; } async function workerReceivesMessage ( { url , promiseId , chat , userId , forwardedFrom , } ) { let primaryMessage ; try { const result = await convert ( url , chat , userId , forwardedFrom ) ; primaryMessage = { ... result , promiseId } ; } catch ( error ) { primaryMessage = { errorString : error . message , promiseId , } ; } process . send ( primaryMessage ) ; } function primaryReceivesMessage ( { textWithTimecodes , duration , promiseId , errorString , } ) { const promiseFunctions = recognitionPromises [ promiseId ] ; if ( promiseFunctions ) { if ( errorString ) { promiseFunctions . rej ( new Error ( errorString ) ) ; } else { promiseFunctions . res ( { textWithTimecodes , duration } ) ; } delete recognitionPromises [ promiseId ] ; } } async function convert ( url , chat , userId ? , forwardedFrom ? ) { let ogaPath ; let flacPath ; try { ogaPath = temp . path ( { suffix : '.oga' } ) ; try { const data = await download ( url ) ; writeFileSync ( ogaPath , data ) ; } catch ( error ) { throw augmentError ( error , 'download url' ) ; } let duration ; try { const result = await flac ( ogaPath ) ; flacPath = result . flacPath ; duration = result . duration ; } catch ( error ) { throw augmentError ( error , 'convert to flac' ) ; } const textWithTimecodes = await getTextFromAudio ( flacPath , chat , duration , ogaPath , userId , forwardedFrom ) ; return { textWithTimecodes , duration , } ; } catch ( error ) { throw augmentError ( error , 'transcribe audio' ) ; } finally { deleteFile ( flacPath ) ; deleteFile ( ogaPath ) ; } }