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

#include <chrono>
#include <cstring>
#include <stdexcept>
#include <string>
#include <vector>
#include <mxl/flow.h>
#include <mxl/time.h>
#include "NodeBase.hpp"
#include "FlowDef.hpp"
#include "V210.hpp"
#include <Processing.NDI.Lib.h>

// RAII wrapper: init NDI, create sender, destroy both on scope exit.
struct NDIContext {
    NDIlib_send_instance_t sender = nullptr;

    explicit NDIContext(const char* ndi_name) {
        if (!NDIlib_is_supported_CPU())
            throw std::runtime_error("CPU not sufficient for NDI");
        if (!NDIlib_initialize())
            throw std::runtime_error("NDI lib init failed");
        NDIlib_send_create_t desc{};
        desc.p_ndi_name = ndi_name;
        sender = NDIlib_send_create(&desc);
        if (!sender) {
            NDIlib_destroy();
            throw std::runtime_error("Cannot create NDI send instance");
        }
    }
    ~NDIContext() {
        NDIlib_send_destroy(sender);
        NDIlib_destroy();
    }
};

class NDIOutNode : public dmf::NodeBase {
    void run() override {
        const auto flow_info = config().at("flow_id");
        const auto flow_id   = flow_info.at("id").get<std::string>();
        const int  width     = flow_info.value("width",   1920);
        const int  height    = flow_info.value("height",  1080);
        const int  fps_num   = flow_info.value("fps_num", 25);
        const int  fps_den   = flow_info.value("fps_den", 1);

        log("flow=%s", flow_id.c_str());

        log("waiting for flow to become active...");
        bool active = false;
        while (!active && dmf::g_running.load(std::memory_order_relaxed)) {
            mxlIsFlowActive(instance(), flow_id.c_str(), &active);
            if (!active) mxlSleepForNs(100'000'000);
        }
        if (!dmf::g_running) return;
        log("flow active — starting read");

        mxlFlowReader reader{};
        mxlStatus st = mxlCreateFlowReader(instance(), flow_id.c_str(), nullptr, &reader);
        if (st != MXL_STATUS_OK) {
            log("mxlCreateFlowReader failed (status=%d)", st);
            return;
        }
        mxlFlowConfigInfo cfg_info{};
        mxlFlowReaderGetConfigInfo(reader, &cfg_info);
        const uint32_t mxl_stride = cfg_info.discrete.sliceSizes[0];

        NDIContext ndi(flow_id.c_str());

        // V210 (10-bit) intermediate and P216 (16-bit) send buffers
        std::vector<uint8_t> buf_10bit(mxl_stride * height);
        std::vector<uint8_t> buf_16bit(width * sizeof(uint16_t) * 2 * height);

        NDIlib_video_frame_v2_t ndi_frame_10bit{};
        ndi_frame_10bit.xres                 = width;
        ndi_frame_10bit.yres                 = height;
        ndi_frame_10bit.FourCC               = static_cast<NDIlib_FourCC_video_type_e>(NDI_LIB_FOURCC('V','2','1','0'));
        ndi_frame_10bit.line_stride_in_bytes = mxl_stride;
        ndi_frame_10bit.p_data               = buf_10bit.data();

        NDIlib_video_frame_v2_t ndi_frame_16bit{};
        ndi_frame_16bit.xres                 = width;
        ndi_frame_16bit.yres                 = height;
        ndi_frame_16bit.frame_rate_N         = fps_num;
        ndi_frame_16bit.frame_rate_D         = fps_den;
        ndi_frame_16bit.line_stride_in_bytes = width * static_cast<int>(sizeof(uint16_t));
        ndi_frame_16bit.p_data               = buf_16bit.data();

        const mxlRational rate = {fps_num, fps_den};
        uint64_t index         = mxlGetCurrentIndex(&rate);
        uint64_t frame_count   = 0;
        uint64_t invalid_count = 0;
        uint64_t late_count    = 0;
        uint64_t ndi_frame_count = 0;
        auto     wall_start    = std::chrono::steady_clock::now();
        auto     last_log_time = wall_start;

        while (dmf::g_running.load(std::memory_order_relaxed)) {
            mxlGrainInfo grain{};
            uint8_t*     buf = nullptr;

            st = mxlFlowReaderGetGrainNonBlocking(reader, index, &grain, &buf);

            if (st == MXL_STATUS_OK) {
                frame_count++;
                if (grain.flags & MXL_GRAIN_FLAG_INVALID) invalid_count++;
                index++;

                if (NDIlib_send_get_no_connections(ndi.sender, 0) > 0) {
                    std::memcpy(ndi_frame_10bit.p_data, buf, mxl_stride * height);
                    NDIlib_util_V210_to_P216(&ndi_frame_10bit, &ndi_frame_16bit);
                    NDIlib_send_send_video_v2(ndi.sender, &ndi_frame_16bit);
                    if (++ndi_frame_count == 1)
                        log("NDI receiver connected");
                } else {
                    ndi_frame_count = 0;
                }

            } else if (st == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) {
                mxlSleepForNs(1'000'000);

            } else if (st == MXL_ERR_OUT_OF_RANGE_TOO_LATE) {
                late_count++;
                mxlFlowRuntimeInfo ri{};
                mxlFlowReaderGetRuntimeInfo(reader, &ri);
                index = ri.headIndex;

            } else {
                log("unexpected status=%d on index=%llu", st, index);
                break;
            }

            auto now = std::chrono::steady_clock::now();
            if (std::chrono::duration<double>(now - last_log_time).count() >= 1.0) {
                const double elapsed = std::chrono::duration<double>(now - wall_start).count();
                log("frames=%llu  invalid=%llu  late=%llu  avg=%.2f fps",
                    frame_count, invalid_count, late_count,
                    static_cast<double>(frame_count) / elapsed);
                last_log_time = now;
            }
        }

        log("stopped — total frames=%llu  invalid=%llu  late=%llu",
            frame_count, invalid_count, late_count);
        mxlReleaseFlowReader(instance(), reader);
    }
};

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