import { clContext as nodenCLContext , OpenCLBuffer } from 'nodencl' ; import { RedioPipe , RedioEnd , nil , isValue , Valve , Spout } from 'redioactive' ; import { Frame , Filterer , filterer } from 'beamcoder' ; import { AudioIO , IoStreamWrite , SampleFormatFloat32 } from 'naudiodon' ; import Koa from 'koa' ; import cors from '@koa/cors' ; import { ConsumerFactory , Consumer } from './consumer' ; import { FromRGBA } from '../process/io' ; import { Writer } from '../process/rgba8' ; import { ConfigParams , VideoFormat , DeviceConfig } from '../config' ; import { ClJobs } from '../clJobQueue' ; interface AudioBuffer { buffer : Buffer ; timestamp : number ; } export class ScreenConsumer implements Consumer { private readonly clContext : nodenCLContext ; private readonly chanID : string ; private readonly params : ConfigParams ; private readonly format : VideoFormat ; private readonly device : DeviceConfig ; private readonly clJobs : ClJobs ; private fromRGBA : FromRGBA | undefined ; private readonly audioOutChannels : number ; private readonly audioTimebase : number [ ] ; private readonly videoTimebase : number [ ] ; private audioOut : IoStreamWrite ; private audFilterer : Filterer | undefined ; private readonly kapp : Koa < Koa . DefaultState , Koa . DefaultContext > ; private readonly lastWeb : Buffer ; private combineAudio : RedioPipe < Frame | RedioEnd > | undefined ; private combineVideo : RedioPipe < OpenCLBuffer | RedioEnd > | undefined ; constructor ( context , chanID , params , format , device , clJobs ) { this . clContext = context ; this . chanID = `${ chanID } screen` ; this . params = params ; this . format = format ; this . device = device ; this . clJobs = clJobs ; this . audioOutChannels = 2 ; this . audioTimebase = [ 1 , this . format . audioSampleRate ] ; this . videoTimebase = [ this . format . duration , this . format . timescale ] ; this . audioOut = AudioIO ( { outOptions : { channelCount : this . audioOutChannels , sampleFormat : SampleFormatFloat32 , sampleRate : this . format . audioSampleRate , closeOnError : false } } ) ; if ( Object . keys ( this . params ) . length > 1 ) console . log ( 'Screen consumer - unused params' , this . params ) ; this . lastWeb = Buffer . alloc ( this . format . width * this . format . height * 4 ) ; this . kapp = new Koa ( ) ; this . kapp . use ( cors ( ) ) ; this . kapp . use ( async ( ctx ) => ( ctx . body = this . lastWeb ) ) ; const server = this . kapp . listen ( 3001 ) ; process . on ( 'SIGHUP' , server . close ) ; } async initialise ( ) { const sampleRate = this . audioTimebase [ 1 ] ; const audInLayout = `${ this . format . audioChannels }c` ; const audOutLayout = `${ this . audioOutChannels }c` ; const samplesPerFrame = ( this . format . audioSampleRate * this . format . duration ) / this . format . timescale ; const outSampleFormat = 'flt' ; this . audFilterer = await filterer ( { filterType : 'audio' , inputParams : [ { name : 'in0:a' , timeBase : this . audioTimebase , sampleRate : sampleRate , sampleFormat : 'fltp' , channelLayout : audInLayout } ] , outputParams : [ { name : 'out0:a' , sampleRate : this . format . audioSampleRate , sampleFormat : outSampleFormat , channelLayout : audOutLayout } ] , filterSpec : `[in0:a] aformat=sample_fmts=${ outSampleFormat }:sample_rates=${ this . format . audioSampleRate }:channel_layouts=${ audOutLayout }, asetnsamples=n=${ samplesPerFrame }:p=1 [out0:a]` } ) ; const width = this . format . width ; const height = this . format . height ; this . fromRGBA = new FromRGBA ( this . clContext , 'sRGB' , new Writer ( width , height , false ) , this . clJobs ) ; await this . fromRGBA . init ( ) ; console . log ( 'Created Screen consumer' ) ; return Promise . resolve ( ) ; } deviceConfig ( ) { return this . device ; } connect ( combineAudio , combineVideo ) { this . combineAudio = combineAudio ; this . combineVideo = combineVideo ; const audFilter = async ( frame ) => { if ( isValue ( frame ) ) { const audFilt = this . audFilterer as Filterer ; const ff = await audFilt . filter ( [ { name : 'in0:a' , frames : [ frame ] } ] ) ; const result = ff [ 0 ] . frames . map ( ( f ) => ( { buffer : f . data [ 0 ] , timestamp : f . pts } ) ) ; return result . length > 0 ? result : nil ; } else { return frame ; } } ; const vidProcess = async ( frame ) => { if ( isValue ( frame ) ) { const fromRGBA = this . fromRGBA as FromRGBA ; const clDests = await fromRGBA . createDests ( ) ; clDests . forEach ( ( d ) => ( d . timestamp = frame . timestamp ) ) ; fromRGBA . processFrame ( this . chanID , frame , clDests , 0 ) ; await this . clJobs . runQueue ( { source : this . chanID , timestamp : frame . timestamp } ) ; return clDests [ 0 ] ; } else { this . clJobs . clearQueue ( this . chanID ) ; return frame ; } } ; const vidSaver = async ( frame ) => { if ( isValue ( frame ) ) { const fromRGBA = this . fromRGBA as FromRGBA ; await fromRGBA . saveFrame ( frame , this . clContext . queue . unload ) ; await this . clContext . waitFinish ( this . clContext . queue . unload ) ; return frame ; } else { return frame ; } } ; const screenSpout = async ( frame ) => { if ( isValue ( frame ) ) { const vidBuf = frame [ 0 ] ; const audBuf = frame [ 1 ] ; if ( ! ( audBuf && isValue ( audBuf ) && vidBuf && isValue ( vidBuf ) ) ) { console . log ( 'One-legged zipper:' , audBuf , vidBuf ) ; if ( vidBuf && isValue ( vidBuf ) ) vidBuf . release ( ) ; return Promise . resolve ( ) ; } const atb = this . audioTimebase ; const ats = ( audBuf . timestamp * atb [ 0 ] ) / atb [ 1 ] ; const vtb = this . videoTimebase ; const vts = ( vidBuf . timestamp * vtb [ 0 ] ) / vtb [ 1 ] ; if ( Math . abs ( ats - vts ) > 0.1 ) console . log ( 'Screen audio and video timestamp mismatch - aud:' , ats , ' vid:' , vts ) ; const write = ( data , cb ) => { if ( ! this . audioOut . write ( data , ( err ) => { if ( err ) console . log ( 'Write Error:' , err ) ; } ) ) { this . audioOut . once ( 'drain' , cb ) ; } else { process . nextTick ( cb ) ; } } ; return new Promise ( ( resolve ) => { vidBuf . copy ( this . lastWeb ) ; write ( audBuf . buffer , ( ) => { vidBuf . release ( ) ; resolve ( ) ; } ) ; } ) ; } else { return Promise . resolve ( ) ; } } ; this . audioOut . start ( ) ; this . combineVideo . valve ( vidProcess ) . valve ( vidSaver ) . zip ( this . combineAudio . valve ( audFilter , { oneToMany : true } ) ) . spout ( screenSpout ) ; } release ( audio , video ) { if ( this . combineAudio !== undefined && this . combineVideo !== undefined ) { audio . unfork ( this . combineAudio ) ; video . unfork ( this . combineVideo ) ; } } } export class ScreenConsumerFactory implements ConsumerFactory < ScreenConsumer > { private readonly clContext : nodenCLContext ; constructor ( clContext ) { this . clContext = clContext ; } createConsumer ( chanID , params , format , device , clJobs ) { const consumer = new ScreenConsumer ( this . clContext , chanID , params , format , device , clJobs ) ; return consumer ; } }