dmf-studio-rnd/nodes/videoin/main.cpp

#include <string>
#include <mxl/flow.h>
#include <mxl/time.h>
#include "V210.hpp"
#include "FlowDef.hpp"
#include "NodeBase.hpp"
#include "VideoReader.hpp"

class VideoInNode : public dmf::NodeBase {
    void run() override {
        const std::string filename = config().value("file", std::string{});
        if (filename.empty()) { log("config missing 'file'"); return; }
        log("file: %s", filename.c_str());

        // Audio pacing assumes steady fixed-size chunks from the demuxer (e.g. 1024-sample AAC
        // packets in MPEG-TS). Containers like MP4/MKV can deliver audio in large bursts, which
        // would require a separate metering buffer to pace correctly. Stick to .ts for now.
        dmf::VideoReader video_reader(filename);
        if (!video_reader.has_video && !video_reader.has_audio) {
            log("no video or audio stream found"); return;
        }

        const bool has_video = config().contains("video_flow_id") && video_reader.has_video;
        mxlFlowWriter     video_writer = nullptr;
        mxlFlowConfigInfo video_cfg{};
        uint32_t   video_stride = 0;
        int        width = 0, height = 0, fps_num = 25, fps_den = 1;
        std::string video_flow_id;

        if (has_video) {
            const auto video_flow_info = config().at("video_flow_id");
            video_flow_id = video_flow_info.at("id").get<std::string>();
            width    = video_reader.video_info.width;
            height   = video_reader.video_info.height;
            fps_num  = video_reader.video_info.fps_num;
            fps_den  = video_reader.video_info.fps_den;

            log("video flow=%s  %dx%d @ %d/%d fps", video_flow_id.c_str(), width, height, fps_num, fps_den);

            bool created = false;
            mxlStatus vst = mxlCreateFlowWriter(
                instance(),
                dmf::make_video_flow_def(video_flow_id, node_id(), width, height, fps_num, fps_den).c_str(),
                "", &video_writer, &video_cfg, &created);
            if (vst != MXL_STATUS_OK) {
                log("mxlCreateFlowWriter failed (%s)", dmf::mxl_status_str(vst));
                return;
            }
            video_stride = video_cfg.discrete.sliceSizes[0];
            log("video stride=%u B/line  grain=%u B  ring=%u grains",
                video_stride, video_stride * static_cast<uint32_t>(height), video_cfg.discrete.grainCount);
        }

        mxlFlowWriter audio_writer = nullptr;
        mxlFlowConfigInfo audio_cfg{};
        int  sample_rate  = video_reader.audio_info.sample_rate;
        int  channels     = video_reader.audio_info.channels;
        int  bit_depth    = 32;
        bool has_audio    = config().contains("audio_flow_id") && video_reader.has_audio;

        int max_audio_samples = 0;
        if (has_audio) {
            const auto audio_flow_info = config().at("audio_flow_id");
            const auto audio_flow_id   = audio_flow_info.at("id").get<std::string>();

            log("audio flow=%s  %d Hz  %dch  %d-bit", audio_flow_id.c_str(), sample_rate, channels, bit_depth);

            bool created = false;
            mxlStatus ast = mxlCreateFlowWriter(
                instance(),
                dmf::make_audio_flow_def(audio_flow_id, node_id(), sample_rate, channels, bit_depth,
                                         fps_num, fps_den).c_str(),
                "", &audio_writer, &audio_cfg, &created);
            if (ast != MXL_STATUS_OK) {
                log("audio mxlCreateFlowWriter failed (%s) — continuing without audio", dmf::mxl_status_str(ast));
                has_audio = false;
            } else {
                log("audio channels=%u  buffer=%u samples",
                    audio_cfg.continuous.channelCount, audio_cfg.continuous.bufferLength);
                size_t max_write = 0;
                mxlFlowWriterGetMaxWriteLengthSamples(audio_writer, &max_write);
                max_audio_samples = static_cast<int>(max_write);
            }
        }
        std::vector<uint8_t> audio_temp(max_audio_samples * channels * sizeof(float));

        const mxlRational video_rate = {fps_num, fps_den};
        const mxlRational audio_rate = {sample_rate, 1};
        uint64_t audio_index = 0;
        if (has_audio) {
            audio_index = mxlGetCurrentIndex(&audio_rate);
        }
        uint64_t video_index = has_video ? mxlGetCurrentIndex(&video_rate) : 0;

        while (dmf::g_running.load(std::memory_order_relaxed)) {
            uint8_t*        video_buf = nullptr;
            mxlGrainInfo    grain{};
            mxlStatus       vst = MXL_ERR_UNSUPPORTED_OPERATION;
            if (has_video) {
                vst = mxlFlowWriterOpenGrain(video_writer, video_index, &grain, &video_buf);
            }

            int out_samples_written = 0;
            dmf::VideoReader::FrameKind frame_kind = video_reader.read_next(
                has_video ? video_buf : nullptr,
                video_stride,
                has_audio ? audio_temp.data() : nullptr,
                max_audio_samples,
                out_samples_written
            );

            if (frame_kind == dmf::VideoReader::FrameKind::None) break;

            if (frame_kind == dmf::VideoReader::FrameKind::Video) {
                if (has_video && vst == MXL_STATUS_OK) {
                    grain.flags       = 0;
                    grain.validSlices = grain.totalSlices;
                    mxlFlowWriterCommitGrain(video_writer, &grain);
                }
                if (has_video) {
                    const uint64_t ns = mxlGetNsUntilIndex(video_index + 1, &video_rate);
                    if (ns > 0 && ns < 2'000'000'000ULL) mxlSleepForNs(ns);
                    video_index = mxlGetCurrentIndex(&video_rate);
                }
            } else if (frame_kind == dmf::VideoReader::FrameKind::Audio) {
                if (has_video && vst == MXL_STATUS_OK) mxlFlowWriterCancelGrain(video_writer);
                // Wait for audio clock to catch up — prevents TOO_EARLY and sample loss
                while (dmf::g_running.load(std::memory_order_relaxed)) {
                    const uint64_t audio_now = mxlGetCurrentIndex(&audio_rate);
                    if (audio_index + static_cast<uint64_t>(out_samples_written) <= audio_now) break;
                    const uint64_t ns = mxlGetNsUntilIndex(audio_index + out_samples_written, &audio_rate);
                    if (ns > 0 && ns < 2'000'000'000ULL) mxlSleepForNs(ns);
                    else break;
                }
                if (!dmf::g_running.load(std::memory_order_relaxed)) break;

                mxlMutableWrappedMultiBufferSlice slice{};
                mxlStatus ast = mxlFlowWriterOpenSamples(audio_writer, audio_index, out_samples_written, &slice);
                if (ast != MXL_STATUS_OK) {
                    log("audio OpenSamples failed (%s) index=%llu — skipping", dmf::mxl_status_str(ast), audio_index);
                    audio_index += out_samples_written;  // advance even on failure — keeps alignment
                    continue;
                }
                // copy per channel from audio_temp into slice
                for (int ch = 0; ch < channels; ch++) {
                    uint8_t* dst0 = static_cast<uint8_t*>(slice.base.fragments[0].pointer)
                        + ch * slice.stride;
                    uint8_t* src = audio_temp.data() + ch * max_audio_samples * sizeof(float);

                    size_t frag0_bytes = slice.base.fragments[0].size;
                    size_t total_bytes = out_samples_written * sizeof(float);

                    if (total_bytes <= frag0_bytes) {
                        std::memcpy(dst0, src, total_bytes);
                    } else {
                        // Wrapped — copy first fragment, then second
                        std::memcpy(dst0, src, frag0_bytes);
                        uint8_t* dst1 = static_cast<uint8_t*>(slice.base.fragments[1].pointer)
                                    + ch * slice.stride;
                        std::memcpy(dst1, src + frag0_bytes, total_bytes - frag0_bytes);
                    }
                }
                mxlFlowWriterCommitSamples(audio_writer);
                audio_index += out_samples_written;
                if (has_video) {
                    const uint64_t current = mxlGetCurrentIndex(&video_rate);
                    if (current > video_index) video_index = current;
                }
            }
        }

        log("stopped at video_index=%llu", static_cast<unsigned long long>(video_index));
        if (has_video) mxlReleaseFlowWriter(instance(), video_writer);
        if (has_audio) mxlReleaseFlowWriter(instance(), audio_writer);
    }
};

int main() {
    VideoInNode node;
    return node.execute();
}