diff --git a/src/producer/elecularProducer.ts b/src/producer/elecularProducer.ts new file mode 100644 index 0000000..416f43a --- /dev/null +++ b/src/producer/elecularProducer.ts @@ -0,0 +1,142 @@ +import redio, { isValue, RedioEnd, RedioPipe, Valve, Funnel, nil, RedioNil, HTTPOptions, isEnd, end, literal } from 'redioactive' +import { VideoFormat, VideoFormats } from '../config' +import { AudioMixFrame } from '../mixer' +import { Producer, ProducerFactory } from './producer' +import { clContext as nodenCLContext, OpenCLBuffer } from 'nodencl' +import { LoadParams } from '../chanLayer' +import { ClJobs } from '../clJobQueue' +import { Reader as rgba8Reader } from '../process/rgba8' +import { ToRGBA } from '../process/io' +import { PackImpl } from '../process/packer' + +// FIXME - bad hack - should come from the stream +const width = 1920 +const height = 1080 + +interface ElecularPacket { + x: number + y: number + width: number + height: number + pts: number + data: Buffer +} + +export class ElecularProducer implements Producer { + private readonly sourceID: string + private readonly params: LoadParams + private readonly clContext: nodenCLContext + private readonly clJobs: ClJobs + private format: VideoFormat + private frame: Buffer = Buffer.alloc(width * height *4) + private vidSource: RedioPipe | undefined + private paused = false + constructor(id: number, params: LoadParams, context: nodenCLContext, clJobs: ClJobs) { + this.sourceID = `P${id} Elecular-HTTP ${params.url} L${params.layer}` + this.params = params + this.clContext = context + this.clJobs = clJobs + this.format = new VideoFormats().get('1080p5000') // default + } + async initialise(consumerFormat: VideoFormat) { + let toRGBA: ToRGBA | null = null + let readImpl: PackImpl = new rgba8Reader(width, height) + toRGBA = new ToRGBA(this.clContext, 'sRGB', '709', readImpl, this.clJobs) + await toRGBA.init() + + const vidLoader: Valve = async (frame) => { + if (isValue(frame)) { + const convert = toRGBA as ToRGBA + const clSources = await convert.createSources() + clSources.forEach((s) => s.timestamp = frame.pts) + await convert.loadFrame(frame.data, clSources, this.clContext.queue.load) + await this.clContext.waitFinish(this.clContext.queue.load) + return clSources + } else { + return frame + } + } + + const waitAndDouble: (x: RedioPipe) => RedioPipe = + (x: RedioPipe) => { + let sentCount = -1 + let lastFrames: { [pts: number]: ElecularPacket } = { } + let lastFramesPTSs: number[] = [] + let outResolver: ((x: ElecularPacket |RedioNil | PromiseLike) => void) | null = null + x.each((frame) => { + lastFrames[frame.pts] = frame + lastFramesPTSs.push(frame.pts) + while (lastFramesPTSs.length > 10) { + delete lastFrames[lastFramesPTSs[0]] + lastFramesPTSs = lastFramesPTSs.slice(1) + } + if (outResolver) { + outResolver(nil) + outResolver = null + } + }) + + const out: Funnel = () => new Promise((resolve, reject) => { + if (lastFramesPTSs.length === 0) { + outResolver = resolve + return + } + if (sentCount === -1) { + sentCount = lastFramesPTSs[0] * 2 + } + let toSend = lastFrames[sentCount / 2 | 0] + if (toSend) { + sentCount++ + resolve(toSend) + } else { + if (lastFramesPTSs.length > 0 && sentCount / 2 > lastFramesPTSs[lastFramesPTSs.length - 1]) { + sentCount++ + resolve(toSend) + } else { + sentCount = -1 + outResolver = resolve + } + } + }) + return redio(out) + } + + const frameMapper = (p: ElecularPacket | RedioEnd) => { + if (isEnd(p)) { + return end + } + p.data.copy(this.frame, width * p.y * 4, 0) + return literal({ + ...p, + data: this.frame + }) + } + + this.vidSource = + waitAndDouble( + redio(this.params.url, { portNumber: 8765, blob: 'data' } as HTTPOptions) + ) + .map(frameMapper) + .valve(vidLoader, { bufferSizeMax: 1, oneToMany: true } ) + . + .pause((frame) => { + if (this.paused && isValue(frame)) (frame as OpenCLBuffer).addRef() + return this.paused + }) + } + getSourceID(): string { return '' } + getFormat(): VideoFormat { return null } + getSourceAudio(): RedioPipe | undefined { return undefined } + getSourceVideo(): RedioPipe | undefined { return fred } + setPaused (pause: boolean): void { } + release(): void { } + +} + +export class ElecularProducerFactory implements ProducerFactory { + constructor(private clContext: nodenCLContext) { } + + createProducer(id: number, params: LoadParams, clJobs: ClJobs): ElecularProducer { + return new ElecularProducer(id, params, this.clContext, clJobs) + } +} \ No newline at end of file diff --git a/src/scratch.txt b/src/scratch.txt new file mode 100644 index 0000000..995001a --- /dev/null +++ b/src/scratch.txt @@ -0,0 +1,4 @@ +PLAY 1-1 ../media/dpp/AS11_DPP_HD_EXAMPLE_1.mxf +PLAY 1-2 ../media/NRK/MINISKI-XX-NO_20200630161932_1_633933.mxf +PLAY 1-2 ../media/NRK/VALDRESTUR-NY-XXXXXX-IN19_20200619133556_15_344695.mxf +PLAY 1-2 ../media/NRK/SMITTESPREDNING-030720-VE_20200703154451_7_491315.mxf