import { ProducerFactory , Producer , InvalidProducerError , ProducerConfig } from './producer' ; import { clContext as nodenCLContext , OpenCLBuffer } from 'nodencl' ; import { Demuxer , demuxer , Decoder , decoder , Filterer , filterer , Stream , Packet , Frame , frame } from 'beamcoder' ; import redio , { RedioPipe , nil , end , isValue , RedioEnd , Generator , Valve , RedioNil } from 'redioactive' ; import { ClJobs } from '../clJobQueue' ; import { LoadParams } from '../chanLayer' ; import { VideoFormat } from '../config' ; import { ToRGBA } from '../process/io' ; import { Reader as nv12Reader } from '../process/nv12' ; import { Reader as yuv422p10Reader } from '../process/yuv422p10' ; import { Reader as yuv422p8Reader } from '../process/yuv422p8' ; import { Reader as yuv420pReader } from '../process/yuv420p' ; import { Reader as v210Reader } from '../process/v210' ; import { Reader as rgba8Reader } from '../process/rgba8' ; import { Reader as bgra8Reader } from '../process/bgra8' ; import Yadif from '../process/yadif' ; import { PackImpl } from '../process/packer' ; import { Mixer } from './mixer' ; import { SourcePipes } from '../routeSource' ; interface AudioChannel { name : string ; frames : Frame [ ] ; } export class FFmpegProducer implements Producer { private readonly sourceID : string ; private readonly loadParams : LoadParams ; private readonly clContext : nodenCLContext ; private readonly clJobs : ClJobs ; private readonly consumerFormat : VideoFormat ; private readonly config : ProducerConfig ; private readonly mixer : Mixer ; private demuxer : Demuxer | null = null ; private audSource : RedioPipe < Frame [ ] | RedioEnd > | undefined ; private vidSource : RedioPipe < OpenCLBuffer | RedioEnd > | undefined ; private srcFormat : VideoFormat | undefined ; private numForks = 0 ; private running = true ; constructor ( id , loadParams , context , clJobs , consumerFormat , config ) { this . sourceID = `P${ id } FFmpeg ${ loadParams . url } L${ loadParams . layer }` ; this . loadParams = loadParams ; this . clContext = context ; this . clJobs = clJobs ; this . consumerFormat = consumerFormat ; this . config = config ; this . mixer = new Mixer ( this . clContext , this . consumerFormat , this . clJobs ) ; } async initialise ( ) { try { this . demuxer = await demuxer ( this . loadParams . url ) ; } catch ( err ) { if ( typeof err === 'string' ) throw new InvalidProducerError ( err ) ; console . log ( `Error in FFmpeg producer initialise: ${ err instanceof Error ? err . message : 'Unknown error' }` ) ; throw err ; } const audioStreams = [ ] ; const videoStreams = [ ] ; const audioIndexes = [ ] ; const videoIndexes = [ ] ; const numVidChannels = 1 ; const audioPackets = [ ] ; const videoPackets = [ ] ; let audioDecoder ; let videoDecoder ; const demuxAudioStreams = this . demuxer . streams . filter ( ( s ) => s . codecpar . codec_type === 'audio' ) ; let astreams = this . loadParams . streams === undefined ? demuxAudioStreams : [ ] ; if ( this . loadParams . streams && this . loadParams . streams . audio ) astreams = demuxAudioStreams . filter ( ( _s , i ) => this . loadParams . streams ?. audio ?. find ( ( loadIndex ) => loadIndex === i ) !== undefined ) ; const demuxVideoStreams = this . demuxer . streams . filter ( ( s ) => s . codecpar . codec_type === 'video' ) ; let vstreams = this . loadParams . streams === undefined ? demuxVideoStreams : [ ] ; if ( this . loadParams . streams && this . loadParams . streams . video ) vstreams = demuxVideoStreams . filter ( ( _s , i ) => this . loadParams . streams ?. video ?. find ( ( loadIndex ) => loadIndex === i ) !== undefined ) ; let monoStreams = true ; astreams . forEach ( ( s ) => { if ( monoStreams ) { s . discard = 'default' ; audioStreams . push ( s ) ; audioIndexes . push ( s . index ) ; if ( ! audioDecoder ) audioDecoder = decoder ( { demuxer : this . demuxer as Demuxer , stream_index : s . index } ) ; monoStreams &&= s . codecpar . channel_layout === 'mono' ; } else { s . discard = 'all' ; } } ) ; vstreams . forEach ( ( s ) => { if ( videoStreams . length < numVidChannels ) { s . discard = 'default' ; videoStreams . push ( s ) ; videoIndexes . push ( s . index ) ; if ( ! videoDecoder ) { videoDecoder = decoder ( Object . assign ( { demuxer : this . demuxer as Demuxer , stream_index : s . index } , this . config . ffmpeg . videoDecoder ) ) ; } } else { s . discard = 'all' ; } } ) ; const primaryIndex = videoIndexes . length ? videoIndexes [ 0 ] : audioIndexes [ 0 ] ; const start = this . loadParams . seek ? this . loadParams . seek : 0 ; if ( start > 0 ) await this . demuxer . seek ( { stream_index : primaryIndex , frame : start } ) ; const maxFrame = this . loadParams . length ? this . loadParams . length : Number . MAX_SAFE_INTEGER ; const loop = this . loadParams . loop ? this . loadParams . loop : false ; let silentFrame = null ; let audFilterer = null ; const audStream = audioStreams [ 0 ] ; let numAudChannels = 0 ; if ( audStream ) { const inParams = audioStreams . map ( ( _s , i ) => { return { name : `in${ i }:a` , timeBase : audStream . time_base , sampleRate : audStream . codecpar . sample_rate , sampleFormat : audStream . codecpar . format , channelLayout : monoStreams ? '1c' : audStream . codecpar . channel_layout } ; } ) ; let filtStr = '' ; if ( monoStreams ) { numAudChannels = audioStreams . length ; for ( let c = 0 ; c < numAudChannels ; ++ c ) filtStr += ( c === 0 ? '' : ';\n' ) + `[in${ c }:a]asetnsamples=n=1024:p=1[out${ c }:a]` ; } else { numAudChannels = audStream . codecpar . channels ; filtStr += `[in${ 0 }:a]asetnsamples=n=1024:p=1, channelsplit=channel_layout=${ numAudChannels }c` ; for ( let s = 0 ; s < numAudChannels ; ++ s ) filtStr += `[c${ s }:a]` ; for ( let s = 0 ; s < numAudChannels ; ++ s ) filtStr += `;\n[c${ s }:a]aformat=channel_layouts=1c[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 : inParams , outputParams : outParams , filterSpec : filtStr } ) ; } else { numAudChannels = 1 ; silentFrame = frame ( { nb_samples : 1024 , format : 's32' , pts : 0 , sample_rate : this . consumerFormat . audioSampleRate , channels : 1 , channel_layout : '1c' , data : [ Buffer . alloc ( 1024 * 4 ) ] } ) ; audFilterer = await filterer ( { filterType : 'audio' , inputParams : [ { name : 'in0:a' , timeBase : [ 1 , this . consumerFormat . audioSampleRate ] , sampleRate : this . consumerFormat . audioSampleRate , sampleFormat : 's32' , channelLayout : '1c' } ] , outputParams : [ { name : 'out0:a' , sampleRate : this . consumerFormat . audioSampleRate , sampleFormat : 'fltp' , channelLayout : '1c' } ] , filterSpec : '[in0:a] asetpts=N/SR/TB [out0:a]' } ) ; } let width = this . consumerFormat . width ; let height = this . consumerFormat . height ; let squareWidth = width ; let squareHeight = height ; let vidTimescale = this . consumerFormat . timescale ; let vidDuration = this . consumerFormat . duration ; let toRGBA = null ; let vidFilterer = null ; let progressive = true ; let yadif = null ; let black = null ; const vidStream = videoStreams [ 0 ] ; if ( vidStream ) { width = vidStream . codecpar . width ; height = vidStream . codecpar . height ; if ( vidStream . codecpar . sample_aspect_ratio [ 0 ] && vidStream . codecpar . sample_aspect_ratio [ 1 ] ) { squareWidth = ( width * vidStream . codecpar . sample_aspect_ratio [ 0 ] ) / vidStream . codecpar . sample_aspect_ratio [ 1 ] ; } squareHeight = height ; const fieldOrder = vidStream . codecpar . field_order ; progressive = fieldOrder === 'progressive' ; if ( ! progressive && vidStream . avg_frame_rate [ 0 ] / vidStream . avg_frame_rate [ 1 ] > 30 ) { console . log ( 'Framerate greater than 30fps - forcing to progressive' ) ; progressive = true ; } vidTimescale = vidStream . avg_frame_rate [ 0 ] * ( progressive ? 1 : 2 ) ; vidDuration = vidStream . avg_frame_rate [ 1 ] ; const tff = fieldOrder === 'unknown' || fieldOrder . split ( ', ' , 2 ) [ 1 ] === 'top displayed first' ; const yadifMode = progressive ? 'send_frame' : 'send_field' ; yadif = new Yadif ( this . clContext , this . clJobs , width , height , { mode : yadifMode , tff : tff } , ! progressive ) ; await yadif . init ( ) ; } else { const numBytesRGBA = width * height * 4 * 4 ; black = await this . clContext . createBuffer ( numBytesRGBA , 'readwrite' , 'coarse' , { width : width , height : height } , 'combinerBlack' ) ; let off = 0 ; const blackFloat = new Float32Array ( numBytesRGBA / 4 ) ; for ( let y = 0 ; y < height ; ++ y ) { for ( let x = 0 ; x < width * 4 ; x += 4 ) { blackFloat [ off + x + 0 ] = 0.0 ; blackFloat [ off + x + 1 ] = 0.0 ; blackFloat [ off + x + 2 ] = 0.0 ; blackFloat [ off + x + 3 ] = 0.0 ; } off += width * 4 ; } await black . hostAccess ( 'writeonly' ) ; Buffer . from ( blackFloat . buffer ) . copy ( black ) ; } let packetFrame = 0 ; const demux = async ( ) => { let packet = end ; if ( this . demuxer && this . running ) { packet = await this . demuxer . read ( ) ; if ( packet && packet . stream_index === primaryIndex ) { packetFrame ++ ; if ( loop && packetFrame === maxFrame ) { await this . demuxer . seek ( { stream_index : primaryIndex , frame : start } ) ; packetFrame = 0 ; } } } else { this . demuxer = null ; } return packet ? packet : end ; } ; const audPacketFilter = async ( packet ) => { if ( isValue ( packet ) ) { if ( ! this . running ) return nil ; if ( audioIndexes . includes ( packet . stream_index ) ) { audioPackets . push ( packet ) ; if ( audioPackets . length === audioStreams . length ) { const result = audioPackets . slice ( 0 ) ; audioPackets . splice ( 0 ) ; return result ; } else return nil ; } else return nil ; } else { return packet ; } } ; const audDecode = async ( packets ) => { if ( isValue ( packets ) ) { if ( ! ( this . running && audioDecoder ) ) return nil ; const decodedFrames = await audioDecoder . decode ( packets ) ; return decodedFrames . frames . map ( ( f , i ) => ( { name : `in${ i }:a` , frames : [ f ] } ) ) ; } else { return packets ; } } ; const audFilter = async ( frames ) => { if ( isValue ( frames ) ) { if ( ! ( this . running && audFilterer ) ) return nil ; const ff = await audFilterer . filter ( frames ) ; if ( ff . reduce ( ( acc , f ) => acc && f . frames && f . frames . length > 0 , true ) ) { const l = ff [ 0 ] . frames . length ; const result = Array . from ( Array ( l ) , ( ) => new Array ( ff . length ) ) ; ff . forEach ( ( chan , c ) => chan . frames . forEach ( ( f , i ) => ( result [ i ] [ c ] = f ) ) ) ; return result ; } else return nil ; } else { return frames ; } } ; const silence = async ( ) => this . running ? [ { name : 'in0:a' , frames : [ silentFrame ] } ] : end ; const vidPacketFilter = async ( packet ) => { if ( isValue ( packet ) ) { if ( ! this . running ) return nil ; if ( videoIndexes . includes ( packet . stream_index ) ) { videoPackets . push ( packet ) ; if ( videoPackets . length === videoStreams . length ) { const result = videoPackets . slice ( 0 ) ; videoPackets . splice ( 0 ) ; return result ; } else return nil ; } else return nil ; } else { return packet ; } } ; const makevidLoader = async ( format ) => { let filterOutputFormat = format ; let readImpl ; let needFilter = false ; switch ( format ) { case 'nv12' : console . log ( 'Using native nv12 loader' ) ; readImpl = new nv12Reader ( width , height ) ; break ; case 'yuv420p' : console . log ( 'Using native yuv420p loader' ) ; readImpl = new yuv420pReader ( width , height ) ; break ; case 'yuv422p' : console . log ( 'Using native yuv422p8 loader' ) ; readImpl = new yuv422p8Reader ( width , height ) ; break ; case 'yuv422p10le' : console . log ( 'Using native yuv422p10 loader' ) ; readImpl = new yuv422p10Reader ( width , height ) ; break ; case 'v210' : console . log ( 'Using native v210 loader' ) ; readImpl = new v210Reader ( width , height ) ; break ; case 'rgba' : console . log ( 'Using native rgba8 loader' ) ; readImpl = new rgba8Reader ( width , height ) ; break ; case 'bgra' : console . log ( 'Using native bgra8 loader' ) ; readImpl = new bgra8Reader ( width , height ) ; break ; default : if ( vidStream . codecpar . format . includes ( 'yuv' ) ) { console . log ( `Non-native loader for ${ vidStream . codecpar . format } - using yuv422p10` ) ; filterOutputFormat = 'yuv422p10le' ; readImpl = new yuv422p10Reader ( width , height ) ; needFilter = true ; } else if ( vidStream . codecpar . format . includes ( 'rgb' ) ) { console . log ( `Non-native loader for ${ vidStream . codecpar . format } - using rgba8` ) ; filterOutputFormat = 'rgba' ; readImpl = new rgba8Reader ( width , height ) ; needFilter = true ; } else throw new Error ( `Unsupported video format '${ vidStream . codecpar . format }' from FFmpeg decoder` ) ; } toRGBA = new ToRGBA ( this . clContext , '709' , '709' , readImpl , this . clJobs ) ; await toRGBA . init ( ) ; const chanTb = [ this . consumerFormat . duration , this . consumerFormat . timescale ] ; if ( needFilter ) { vidFilterer = await filterer ( { filterType : 'video' , inputParams : [ { timeBase : vidStream . time_base , width : width , height : height , pixelFormat : vidStream . codecpar . format , pixelAspect : vidStream . codecpar . sample_aspect_ratio } ] , outputParams : [ { pixelFormat : filterOutputFormat } ] , filterSpec : `fps=fps=${ chanTb [ 1 ] / ( progressive ? 1 : 2 ) }/${ chanTb [ 0 ] }` } ) ; } } ; const vidDecode = async ( packets ) => { if ( isValue ( packets ) ) { if ( ! ( this . running && videoDecoder ) ) return nil ; let result = nil ; const frm = await videoDecoder . decode ( packets ) ; if ( frm . frames . length > 0 ) { if ( ! toRGBA ) { const decodedFormat = videoDecoder . sw_pix_fmt ? videoDecoder . sw_pix_fmt : vidStream . codecpar . format ; await makevidLoader ( decodedFormat ) ; } result = frm . frames ; } return result ; } else { return packets ; } } ; const vidFilter = async ( decFrame ) => { if ( isValue ( decFrame ) ) { if ( ! vidFilterer ) return decFrame ; if ( ! this . running ) return nil ; const ff = await vidFilterer . filter ( [ decFrame ] ) ; if ( ! ff [ 0 ] ) return nil ; return ff [ 0 ] . frames . length > 0 ? ff [ 0 ] . frames : nil ; } else { return decFrame ; } } ; const vidLoader = async ( frame ) => { if ( isValue ( frame ) ) { if ( ! this . running ) return nil ; const convert = toRGBA as ToRGBA ; const clSources = await convert . createSources ( ) ; clSources . forEach ( ( s ) => { s . timestamp = progressive ? frame . pts : frame . pts * 2 ; } ) ; await convert . loadFrame ( frame . data , clSources , this . clContext . queue . load ) ; await this . clContext . waitFinish ( this . clContext . queue . load ) ; return clSources ; } else { return frame ; } } ; const vidProcess = async ( clSources ) => { if ( isValue ( clSources ) ) { if ( ! this . running ) { clSources . forEach ( ( s ) => s . release ( ) ) ; return nil ; } const convert = toRGBA as ToRGBA ; const clDest = await convert . createDest ( { width : width , height : height } ) ; clDest . timestamp = clSources [ 0 ] . timestamp ; convert . processFrame ( this . sourceID , clSources , clDest ) ; return clDest ; } else { toRGBA ?. finish ( ) ; toRGBA = null ; return clSources ; } } ; let curFrame = 0 ; const vidDeint = async ( frame ) => { if ( isValue ( frame ) ) { if ( ! this . running ) { frame . release ( ) ; return nil ; } const yadifDests = [ ] ; await yadif ?. processFrame ( frame , yadifDests , this . sourceID ) ; if ( ! loop && maxFrame === curFrame ) this . release ( ) ; curFrame += yadifDests . length ; yadifDests . forEach ( ( d ) => { for ( let f = 0 ; f < this . numForks ; ++ f ) d . addRef ( ) ; } ) ; return yadifDests . length > 0 ? yadifDests : nil ; } else { yadif ?. release ( ) ; yadif = null ; this . running = false ; return frame ; } } ; let blackCurFrame = 0 ; const blackPipe = redio ( async ( ) => { if ( this . running ) { if ( ! loop && maxFrame === blackCurFrame ) this . release ( ) ; if ( black ) black . timestamp = blackCurFrame ; blackCurFrame ++ ; black ?. addRef ( ) ; return black ; } else return end ; } , { bufferSizeMax : 1 } ) ; this . srcFormat = { name : 'ffmpeg' , fields : 1 , width : width , height : height , squareWidth : squareWidth , squareHeight : squareHeight , timescale : vidTimescale , duration : vidDuration , audioSampleRate : 48000 , audioChannels : numAudChannels } ; const ffPackets = redio ( demux , { bufferSizeMax : this . demuxer . streams . length * 2 } ) ; if ( audioStreams . length ) { this . audSource = ffPackets . fork ( { bufferSizeMax : 10 } ) . valve ( audPacketFilter ) . valve ( audDecode , { bufferSizeMax : 2 } ) . valve ( audFilter , { oneToMany : true } ) ; } else { this . audSource = redio ( silence , { bufferSizeMax : 2 } ) . valve ( audFilter , { oneToMany : true } ) ; } if ( videoStreams . length ) { this . vidSource = ffPackets . fork ( { bufferSizeMax : 10 } ) . valve ( vidPacketFilter ) . valve ( vidDecode , { oneToMany : true , bufferSizeMax : 2 } ) . valve ( vidFilter , { oneToMany : true } ) . valve ( vidLoader , { bufferSizeMax : 1 } ) . valve ( vidProcess ) . valve ( vidDeint , { oneToMany : true } ) ; } else { this . vidSource = blackPipe ; } await this . mixer . init ( this . sourceID , this . audSource . fork ( ) , this . vidSource . fork ( ) , this . srcFormat ) ; console . log ( `Created FFmpeg producer for path ${ this . loadParams . url }` ) ; } async getSourcePipes ( ) { if ( ! ( this . audSource && this . vidSource && this . srcFormat ) ) throw new Error ( `Route producer failed to find source pipes for route` ) ; this . numForks ++ ; return Promise . resolve ( { audio : this . audSource , video : this . vidSource , format : this . srcFormat , release : ( ) => this . numForks -- } ) ; } getMixer ( ) { return this . mixer ; } setPaused ( pause ) { this . mixer . setPaused ( pause ) ; } release ( ) { this . running = false ; this . mixer . release ( ) ; } } export class FFmpegProducerFactory implements ProducerFactory < FFmpegProducer > { private clContext : nodenCLContext ; constructor ( clContext ) { this . clContext = clContext ; } createProducer ( id , loadParams , clJobs , consumerFormat , config ) { return new FFmpegProducer ( id , loadParams , this . clContext , clJobs , consumerFormat , config ) ; } }