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

extern "C" {
    #include <libavformat/avformat.h>
    #include <libavcodec/avcodec.h>
    #include <libavutil/avutil.h>
    #include <libswscale/swscale.h>
    #include <libavutil/imgutils.h>
}

#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());

        dmf::VideoReader video_reader(filename);
        if (!video_reader.have_video) { log("no video stream found"); return; }

        const auto video_flow_info = config().at("video_flow_id");
        const auto video_flow_id   = video_flow_info.at("id").get<std::string>();
        const int  width    = video_flow_info.value("width",   video_reader.video_info.width);
        const int  height   = video_flow_info.value("height",  video_reader.video_info.height);
        const int  fps_num  = video_flow_info.value("fps_num", video_reader.video_info.fps_num);
        const int  fps_den  = video_flow_info.value("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);

        mxlFlowWriter    video_writer = nullptr;
        mxlFlowConfigInfo video_cfg   = {};
        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;
        }
        const uint32_t 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{};
        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);

            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};
        uint64_t audio_index = 0;
        if (has_audio) {
            const mxlRational audio_rate = {sample_rate, 1};
            audio_index = mxlGetCurrentIndex(&audio_rate);
        }
        uint64_t video_index = mxlGetCurrentIndex(&video_rate);

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

            int out_samples_written = 0;
            dmf::VideoReader::FrameKind frame_kind = video_reader.get_next_frame(
                video_buf,
                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 (vst == MXL_STATUS_OK) {
                    grain.flags       = 0;
                    grain.validSlices = grain.totalSlices;
                    mxlFlowWriterCommitGrain(video_writer, &grain);
                }
                const uint64_t ns = mxlGetNsUntilIndex(video_index + 1, &video_rate);
                if (ns > 0 && ns < 2'000'000'000ULL) mxlSleepForNs(ns);
                video_index++;
            } else if (frame_kind == dmf::VideoReader::FrameKind::Audio) {
                mxlFlowWriterCancelGrain(video_writer);
                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", dmf::mxl_status_str(ast), audio_index);
                    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;
            }
        }

        log("stopped at video_index=%llu", video_index);
        mxlReleaseFlowWriter(instance(), video_writer);
        mxlReleaseFlowWriter(instance(), audio_writer);
    }
};

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