import { EventEmitter } from 'events' ; import { clContext as nodenCLContext } from 'nodencl' ; import { Layer } from './layer' ; import { RedioPipe , RedioEnd , isValue , isEnd , Valve , nil , end } from 'redioactive' ; import { OpenCLBuffer } from 'nodencl' ; import { AudioInputParam , filterer , Filterer , Frame } from 'beamcoder' ; import { VideoFormat } from './config' ; import { ClJobs } from './clJobQueue' ; import ImageProcess from './process/imageProcess' ; import Combine from './process/combine' ; import { Silence , Black } from './blackSilence' ; import { SourcePipes , RouteSource } from './routeSource' ; import { Consumer } from './consumer/consumer' ; export class CombineLayer { private readonly audioPipe : RedioPipe < Frame | RedioEnd > ; private readonly videoPipe : RedioPipe < OpenCLBuffer | RedioEnd > ; private readonly endEvent : EventEmitter ; private audioState : 'start' | 'run' | 'end' = 'start' ; private videoState : 'start' | 'run' | 'end' = 'start' ; constructor ( layer , audPipe , vidPipe ) { this . audioPipe = audPipe ; this . videoPipe = vidPipe ; this . endEvent = layer . getEndEvent ( ) ; } getAudioPipe ( ) { return this . audioPipe ; } getVideoPipe ( ) { return this . videoPipe ; } checkAudio ( frame ) { let result = true ; if ( isValue ( frame ) ) { if ( this . audioState === 'start' ) this . audioState = 'run' ; } else { if ( this . audioState === 'run' ) { this . audioState = 'end' ; if ( this . audioState === 'end' && this . videoState === 'end' ) this . endEvent . emit ( 'end' ) ; } result = false ; } return result ; } checkVideo ( frame ) { let result = true ; if ( isValue ( frame ) ) { if ( this . videoState === 'start' ) this . videoState = 'run' ; } else { if ( this . videoState === 'run' ) { this . videoState = 'end' ; if ( this . audioState === 'end' && this . videoState === 'end' ) this . endEvent . emit ( 'end' ) ; } result = false ; } return result ; } } export class Combiner implements RouteSource { private readonly clContext : nodenCLContext ; private readonly chanID : string ; private readonly consumerFormat : VideoFormat ; private readonly clJobs : ClJobs ; private lastNumAudLayers = 0 ; private lastNumVidLayers = 0 ; private numConsumers = 0 ; private audCombiner : Filterer | undefined ; private vidCombiner : ImageProcess | undefined ; private audioPipe : RedioPipe < Frame | RedioEnd > | undefined ; private videoPipe : RedioPipe < OpenCLBuffer | RedioEnd > | undefined ; private combineLayers : CombineLayer [ ] = [ ] ; private audLayerPipes : RedioPipe < Frame | RedioEnd > [ ] = [ ] ; private vidLayerPipes : RedioPipe < OpenCLBuffer | RedioEnd > [ ] = [ ] ; private audRoutePipe : RedioPipe < Frame [ ] | RedioEnd > | undefined ; private vidTimestamp = 0 ; private numForks = 0 ; constructor ( clContext , chanID , consumerFormat , clJobs ) { this . clContext = clContext ; this . chanID = `${ chanID } combine` ; this . consumerFormat = consumerFormat ; this . clJobs = clJobs ; } async initialise ( ) { const silence = new Silence ( this . consumerFormat ) ; const silencePipe = await silence . initialise ( ) ; const black = new Black ( this . clContext , this . consumerFormat , this . chanID ) ; const blackPipe = await black . initialise ( ) ; const audEndValve = async ( frames ) => { if ( isValue ( frames ) ) { return frames . filter ( ( f , i ) => i > 0 ? this . combineLayers . length > i - 1 ? this . combineLayers [ i - 1 ] . checkAudio ( f ) : false : true ) as [ Frame | RedioEnd , ... ( Frame | RedioEnd ) [ ] ] ; } else { return frames ; } } ; const combineAudValve = async ( frames ) => { if ( isValue ( frames ) ) { const numLayers = frames . length - 1 ; const layerFrames = frames . slice ( 1 ) as Frame [ ] ; const doFilter = layerFrames . reduce ( ( acc , f ) => acc && isValue ( f ) , true ) ; const numCombineLayers = numLayers < 2 ? 0 : numLayers ; if ( numCombineLayers && this . lastNumAudLayers !== numCombineLayers ) { await this . makeAudCombiner ( numCombineLayers ) ; this . lastNumAudLayers = numCombineLayers ; } const srcFrames = frames as Frame [ ] ; if ( ! isValue ( frames [ 0 ] ) ) return end ; const refFrame = srcFrames [ 0 ] ; if ( numLayers === 0 ) { return srcFrames [ 0 ] ; } else if ( numLayers === 1 ) { if ( isValue ( srcFrames [ 1 ] ) ) srcFrames [ 1 ] . pts = refFrame . pts ; return srcFrames [ 1 ] ; } else if ( doFilter && this . audCombiner ) { const filterFrames = layerFrames . map ( ( f , i ) => { f . pts = refFrame . pts ; return { name : `in${ i }:a` , frames : [ f ] } ; } ) ; const ff = await this . audCombiner . filter ( filterFrames ) ; return ff [ 0 ] . frames . length > 0 ? ff [ 0 ] . frames : nil ; } else { return end ; } } else { this . audCombiner = undefined ; silence . release ( ) ; return frames ; } } ; const vidEndValve = async ( frames ) => { if ( isValue ( frames ) ) { return frames . filter ( ( f , i ) => i > 0 ? this . combineLayers . length > i - 1 ? this . combineLayers [ i - 1 ] . checkVideo ( f ) : false : true ) as [ OpenCLBuffer | RedioEnd , ... ( OpenCLBuffer | RedioEnd ) [ ] ] ; } else { return frames ; } } ; const combineVidValve = async ( frames ) => { let result = end ; if ( isValue ( frames ) && isValue ( frames [ 0 ] ) ) { const layerFrames = frames . slice ( 1 ) ; const numLayers = layerFrames . length ; const timestamp = this . vidTimestamp ++ ; const numCombineLayers = numLayers < 2 ? 0 : numLayers ; if ( numCombineLayers && this . lastNumVidLayers !== numCombineLayers ) { await this . makeVidCombiner ( numCombineLayers ) ; this . lastNumVidLayers = numCombineLayers ; } if ( numLayers === 0 ) { frames [ 0 ] . timestamp = timestamp ; frames [ 0 ] . addRef ( ) ; result = frames [ 0 ] ; } else if ( numLayers === 1 ) { if ( ! isEnd ( frames [ 1 ] ) ) { frames [ 1 ] . timestamp = timestamp ; frames [ 1 ] . addRef ( ) ; } result = frames [ 1 ] ; } else if ( layerFrames . reduce ( ( acc , f ) => acc && isValue ( f ) , true ) ) { const combineDest = await this . clContext . createBuffer ( this . consumerFormat . width * this . consumerFormat . height * 4 * 4 , 'readwrite' , 'coarse' , { width : this . consumerFormat . width , height : this . consumerFormat . height } , 'combine' ) ; combineDest . timestamp = timestamp ; await this . vidCombiner ?. run ( { inputs : layerFrames , output : combineDest } , { source : this . chanID , timestamp : timestamp } , ( ) => { } ) ; await this . clJobs . runQueue ( { source : this . chanID , timestamp : timestamp } ) ; result = combineDest ; } if ( isValue ( result ) ) for ( let d = 1 ; d < this . numConsumers + this . numForks ; ++ d ) result . addRef ( ) ; frames . forEach ( ( f ) => ( isValue ( f ) ? f . release ( ) : { } ) ) ; } else { if ( this . vidCombiner ) { this . clJobs . clearQueue ( this . chanID ) ; black . release ( ) ; this . vidCombiner = undefined ; } } return result ; } ; this . audioPipe = silencePipe . zipEach ( this . audLayerPipes ) . valve ( audEndValve ) . valve ( combineAudValve , { oneToMany : true } ) ; this . videoPipe = blackPipe . zipEach ( this . vidLayerPipes ) . valve ( vidEndValve ) . valve ( combineVidValve ) ; } async makeAudCombiner ( numLayers ) { const sampleRate = this . consumerFormat . audioSampleRate ; const numAudChannels = this . consumerFormat . audioChannels ; const audLayout = `${ numAudChannels }c` ; const inParams = [ ] ; let inStr = '' ; const filtLayers = numLayers > 0 ? numLayers : 1 ; for ( let i = 0 ; i < filtLayers ; i ++ ) { inStr += `[in${ i }:a]` ; inParams . push ( { name : `in${ i }:a` , timeBase : [ 1 , sampleRate ] , sampleRate : sampleRate , sampleFormat : 'fltp' , channelLayout : `${ numAudChannels }c` } ) ; } this . audCombiner = await filterer ( { filterType : 'audio' , inputParams : inParams , outputParams : [ { name : 'out0:a' , sampleRate : sampleRate , sampleFormat : 'fltp' , channelLayout : audLayout } ] , filterSpec : `${ inStr }amix=inputs=${ filtLayers }:duration=shortest[out0:a]` } ) ; } async makeVidCombiner ( numLayers ) { this . vidCombiner = new ImageProcess ( this . clContext , new Combine ( numLayers , this . consumerFormat . width , this . consumerFormat . height ) , this . clJobs ) ; await this . vidCombiner . init ( ) ; } getLayers ( ) { return this . combineLayers ; } updateLayers ( layers ) { this . combineLayers = layers . slice ( 0 ) ; this . audLayerPipes . splice ( 0 ) ; this . vidLayerPipes . splice ( 0 ) ; layers . forEach ( ( l ) => { this . audLayerPipes . push ( l . getAudioPipe ( ) ) ; this . vidLayerPipes . push ( l . getVideoPipe ( ) ) ; } ) ; } connect ( consumer ) { if ( ! ( this . audioPipe !== undefined && this . videoPipe !== undefined ) ) { throw new Error ( 'Failed to get combiner connection pipes' ) ; } consumer . connect ( this . audioPipe . fork ( ) , this . videoPipe . fork ( ) ) ; this . numConsumers ++ ; } release ( consumer ) { if ( ! ( this . audioPipe !== undefined && this . videoPipe !== undefined ) ) { throw new Error ( 'Failed to get combiner connection pipes' ) ; } consumer . release ( this . audioPipe , this . videoPipe ) ; this . numConsumers -- ; } async getSourcePipes ( ) { if ( ! ( this . audioPipe && this . videoPipe && this . consumerFormat ) ) throw new Error ( `Combiner failed to find source pipes for route` ) ; if ( this . numForks === 0 ) { let audFilterer = null ; let filtStr = '' ; const numAudChannels = this . consumerFormat . audioChannels ; const sampleRate = this . consumerFormat . audioSampleRate ; filtStr += `[in${ 0 }:a]channelsplit=channel_layout=${ numAudChannels }c` ; for ( let s = 0 ; s < numAudChannels ; ++ s ) filtStr += `[out${ s }:a]` ; const outParams = [ ] ; for ( let s = 0 ; s < numAudChannels ; ++ s ) { outParams . push ( { name : `out${ s }:a` , sampleRate : this . consumerFormat . audioSampleRate , sampleFormat : 'fltp' , channelLayout : '1c' } ) ; } audFilterer = await filterer ( { filterType : 'audio' , inputParams : [ { name : 'in0:a' , timeBase : [ 1 , sampleRate ] , sampleRate : sampleRate , sampleFormat : 'fltp' , channelLayout : `${ numAudChannels }c` } ] , outputParams : outParams , filterSpec : filtStr } ) ; const audFilter = async ( frame ) => { if ( isValue ( frame ) ) { if ( ! audFilterer ) return nil ; const ff = await audFilterer . filter ( [ { name : 'in0:a' , frames : [ frame ] } ] ) ; if ( ff . reduce ( ( acc , f ) => acc && f . frames && f . frames . length > 0 , true ) ) { return ff . map ( ( f ) => f . frames [ 0 ] ) ; } else return nil ; } else { return frame ; } } ; this . audRoutePipe = this . audioPipe . fork ( ) . valve ( audFilter ) ; } this . numForks ++ ; if ( ! this . audRoutePipe ) throw new Error ( `Combiner failed to create audio filter for route` ) ; return { audio : this . audRoutePipe , video : this . videoPipe , format : this . consumerFormat , release : ( ) => this . numForks -- } ; } getAudioPipe ( ) { return this . audioPipe ; } getVideoPipe ( ) { return this . videoPipe ; } }