#include #include #include #include #include #include #include #include "NodeBase.hpp" #include "FlowDef.hpp" #include "V210.hpp" #include // 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(); 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 buf_10bit(mxl_stride * height); std::vector 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(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(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(now - last_log_time).count() >= 1.0) { const double elapsed = std::chrono::duration(now - wall_start).count(); log("frames=%llu invalid=%llu late=%llu avg=%.2f fps", frame_count, invalid_count, late_count, static_cast(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(); }