#include #include #include #include "NodeBase.hpp" #include "FlowDef.hpp" #include "V210.hpp" #include "NDIHelper.hpp" #include #include class NDIInNode : public dmf::NodeBase { void run() override { const uint32_t source_num = config().value("source_num", 0); dmf::NDIHelper ndi_helper; try { std::vector ndi_sources; ndi_helper.find_sources(&ndi_sources, 5000); log("Available NDI sources:"); for (const auto& source_name : ndi_sources) log(" %s", source_name.c_str()); ndi_helper.select_source(source_num); ndi_helper.get_source_info(source_num); } catch (const std::runtime_error& e) { log("Error: %s", e.what()); return; } const auto flow_info = config().at("flow_id"); const auto flow_id = flow_info.at("id").get(); const int width = flow_info.value("width", ndi_helper.xres); const int height = flow_info.value("height", ndi_helper.yres); const int fps_num = flow_info.value("fps_num", ndi_helper.frame_N); const int fps_den = flow_info.value("fps_den", ndi_helper.frame_D); log("flow=%s %dx%d @ %d/%d fps", flow_id.c_str(), width, height, fps_num, fps_den); const std::string flow_def = dmf::make_video_flow_def(flow_id, node_id(), width, height, fps_num, fps_den); mxlFlowWriter writer{}; mxlFlowConfigInfo cfg_info{}; bool created = false; mxlStatus st = mxlCreateFlowWriter( instance(), flow_def.c_str(), nullptr, &writer, &cfg_info, &created); if (st != MXL_STATUS_OK) { log("mxlCreateFlowWriter failed (status=%d)", st); return; } const uint32_t stride = cfg_info.discrete.sliceSizes[0]; log("stride=%u B/line grain=%u B ring=%u grains", stride, stride * static_cast(height), cfg_info.discrete.grainCount); const mxlRational rate = {fps_num, fps_den}; uint64_t index = mxlGetCurrentIndex(&rate); log("start index=%llu", index); // because NDI can drop to 1 FPS for static frames, even if source is 29.97p const size_t frame_bytes = stride * height; uint8_t* latest_buffer = (uint8_t*)malloc(frame_bytes); bool last_ndi_frame_valid = false; while (dmf::g_running.load(std::memory_order_relaxed)) { try { if (ndi_helper.getV210_video_frame(source_num, latest_buffer, stride)) last_ndi_frame_valid = true; } catch (const std::runtime_error& e) { log("NDI error: %s — stopping", e.what()); break; } uint8_t* buf = nullptr; mxlGrainInfo grain{}; st = mxlFlowWriterOpenGrain(writer, index, &grain, &buf); if (st != MXL_STATUS_OK) { log("OpenGrain failed (status=%d), skipping index=%llu", st, index); index++; continue; } if (last_ndi_frame_valid) { std::memcpy(buf, latest_buffer, frame_bytes); grain.flags = 0; } else { grain.flags = MXL_GRAIN_FLAG_INVALID; } grain.validSlices = grain.totalSlices; // mark grain complete so readers can consume it mxlFlowWriterCommitGrain(writer, &grain); const uint64_t ns = mxlGetNsUntilIndex(index + 1, &rate); if (ns > 0 && ns < 2'000'000'000ULL) mxlSleepForNs(ns); index++; } log("stopped at index=%llu", index); free(latest_buffer); mxlReleaseFlowWriter(instance(), writer); } }; int main() { NDIInNode node; return node.execute(); }