itten/dmf-studio-rnd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Feature/ndi out node #2

Merged
itten merged 6 commits from feature/ndi-out-node into main 2026-07-03 10:13:26 +03:00
Conversation 0 Commits 6 Files Changed 5
+253 -135
5 changed files with 253 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.vscode/c_cpp_properties.json
Vendored
+1
View File
@@ -4,6 +4,7 @@
"name": "Linux",
"includePath": [
"${workspaceFolder}/**",
"${workspaceFolder}/shared",
"${HOME}/SDK/NDI/include"
],
"defines": [],
nodes/ndiin/main.cpp
+5 -2
View File
@@ -81,10 +81,13 @@ class NDIInNode : public dmf::NodeBase {
// --- main loop ---
const mxlRational video_rate = {fps_num, fps_den};
const mxlRational audio_rate = {sample_rate, 1};
uint64_t video_index = mxlGetCurrentIndex(&video_rate);
uint64_t audio_index = has_audio ? mxlGetCurrentIndex(&audio_rate) : 0;
uint64_t audio_index = 0;
if (has_audio) {
const mxlRational audio_rate = {sample_rate, 1};
audio_index = mxlGetCurrentIndex(&audio_rate);
}
log("start video_index=%llu", video_index);
std::vector<uint8_t> latest_video(video_stride * height);
nodes/ndiout/main.cpp
+161 -52
View File
@@ -6,8 +6,6 @@
#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.
@@ -35,14 +33,26 @@ struct NDIContext {
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);
// --- video flow (optional) ---
bool has_video = config().contains("flow_id");
log("flow=%s", flow_id.c_str());
int width = 1920;
int height = 1080;
int fps_num = 25;
int fps_den = 1;
std::string flow_id;
mxlFlowReader video_reader{};
uint32_t video_stride = 0;
if (has_video) {
const auto flow_info = config().at("flow_id");
flow_id = flow_info.at("id").get<std::string>();
width = flow_info.value("width", 1920);
height = flow_info.value("height", 1080);
fps_num = flow_info.value("fps_num", 25);
fps_den = flow_info.value("fps_den", 1);
log("video flow=%s %dx%d @ %d/%d fps", flow_id.c_str(), width, height, fps_num, fps_den);
log("waiting for flow to become active...");
bool active = false;
@@ -53,39 +63,85 @@ class NDIOutNode : public dmf::NodeBase {
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 video_cfg{};
mxlStatus vst = mxlCreateFlowReader(instance(), flow_id.c_str(), nullptr, &video_reader);
if (vst != MXL_STATUS_OK) { log("video mxlCreateFlowReader failed (status=%d)", vst); return; }
mxlFlowReaderGetConfigInfo(video_reader, &video_cfg);
video_stride = video_cfg.discrete.sliceSizes[0];
}
mxlFlowConfigInfo cfg_info{};
mxlFlowReaderGetConfigInfo(reader, &cfg_info);
const uint32_t mxl_stride = cfg_info.discrete.sliceSizes[0];
NDIContext ndi(flow_id.c_str());
// --- audio flow (optional) ---
mxlFlowReader audio_reader{};
mxlFlowConfigInfo audio_cfg{};
int sample_rate = 0;
int channels = 0;
int samples_per_frame = 0;
bool has_audio = config().contains("audio_flow_id");
// 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);
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>();
sample_rate = audio_flow_info.value("sample_rate", 48000);
channels = audio_flow_info.value("channels", 2);
samples_per_frame = sample_rate * fps_den / fps_num;
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();
log("audio flow=%s %d Hz %dch %d samples/frame",
audio_flow_id.c_str(), sample_rate, channels, samples_per_frame);
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();
mxlStatus ast = mxlCreateFlowReader(instance(), audio_flow_id.c_str(), nullptr, &audio_reader);
if (ast != MXL_STATUS_OK) {
log("audio mxlCreateFlowReader failed (status=%d) — continuing without audio", ast);
has_audio = false;
} else {
mxlFlowReaderGetConfigInfo(audio_reader, &audio_cfg);
log("audio channels=%u buffer=%u samples",
audio_cfg.continuous.channelCount, audio_cfg.continuous.bufferLength);
}
}
const mxlRational rate = {fps_num, fps_den};
uint64_t index = mxlGetCurrentIndex(&rate);
if (!has_video && !has_audio) { log("no flows configured — exiting"); return; }
NDIContext ndi(node_id().c_str());
// Video buffers and NDI frames (only when has_video)
std::vector<uint8_t> v210_buf, p216_buf;
NDIlib_video_frame_v2_t v210_frame{}, p216_frame{};
if (has_video) {
v210_buf.resize(video_stride * height);
p216_buf.resize(width * sizeof(uint16_t) * 2 * height);
v210_frame.xres = width;
v210_frame.yres = height;
v210_frame.frame_rate_N = fps_num;
v210_frame.frame_rate_D = fps_den;
v210_frame.FourCC = static_cast<NDIlib_FourCC_video_type_e>(NDI_LIB_FOURCC('V','2','1','0'));
v210_frame.line_stride_in_bytes = video_stride;
v210_frame.p_data = v210_buf.data();
p216_frame.xres = width;
p216_frame.yres = height;
p216_frame.FourCC = NDIlib_FourCC_video_type_P216;
p216_frame.frame_rate_N = fps_num;
p216_frame.frame_rate_D = fps_den;
p216_frame.line_stride_in_bytes = width * static_cast<int>(sizeof(uint16_t));
p216_frame.p_data = p216_buf.data();
}
// Audio buffer and NDI frame (only when has_audio)
std::vector<float> audio_planar(static_cast<size_t>(channels) * samples_per_frame);
NDIlib_audio_frame_v3_t ndi_audio{};
if (has_audio) {
ndi_audio.sample_rate = sample_rate;
ndi_audio.no_channels = channels;
ndi_audio.no_samples = samples_per_frame;
ndi_audio.FourCC = NDIlib_FourCC_audio_type_FLTP;
ndi_audio.channel_stride_in_bytes = samples_per_frame * sizeof(float);
ndi_audio.p_data = reinterpret_cast<uint8_t*>(audio_planar.data());
}
// --- main loop ---
uint64_t video_index = 0;
uint64_t audio_index = 0;
uint64_t frame_count = 0;
uint64_t invalid_count = 0;
uint64_t late_count = 0;
@@ -93,38 +149,82 @@ class NDIOutNode : public dmf::NodeBase {
auto wall_start = std::chrono::steady_clock::now();
auto last_log_time = wall_start;
if (has_video) {
const mxlRational video_rate = {fps_num, fps_den};
video_index = mxlGetCurrentIndex(&video_rate);
}
if (has_audio) {
const mxlRational audio_rate = {sample_rate, 1};
audio_index = mxlGetCurrentIndex(&audio_rate);
}
while (dmf::g_running.load(std::memory_order_relaxed)) {
mxlGrainInfo grain{};
uint8_t* buf = nullptr;
// --- audio: non-blocking, one chunk per video frame (or free-running) ---
bool audio_advanced = false;
if (has_audio) {
mxlWrappedMultiBufferSlice audio_slices{};
mxlStatus ast = mxlFlowReaderGetSamplesNonBlocking(
audio_reader, audio_index, samples_per_frame, &audio_slices);
if (ast == MXL_STATUS_OK) {
const size_t frag0 = audio_slices.base.fragments[0].size / sizeof(float);
const size_t frag1 = audio_slices.base.fragments[1].size / sizeof(float);
for (int c = 0; c < channels; ++c) {
float* dst = audio_planar.data() + c * samples_per_frame;
const auto* src0 = reinterpret_cast<const float*>(
static_cast<const uint8_t*>(audio_slices.base.fragments[0].pointer)
+ c * audio_slices.stride);
std::memcpy(dst, src0, frag0 * sizeof(float));
if (frag1 > 0) {
const auto* src1 = reinterpret_cast<const float*>(
static_cast<const uint8_t*>(audio_slices.base.fragments[1].pointer)
+ c * audio_slices.stride);
std::memcpy(dst + frag0, src1, frag1 * sizeof(float));
}
}
NDIlib_send_send_audio_v3(ndi.sender, &ndi_audio);
audio_index += samples_per_frame;
audio_advanced = true;
} else if (ast == MXL_ERR_OUT_OF_RANGE_TOO_LATE) {
mxlFlowRuntimeInfo ari{};
mxlFlowReaderGetRuntimeInfo(audio_reader, &ari);
audio_index = ari.headIndex;
}
}
st = mxlFlowReaderGetGrainNonBlocking(reader, index, &grain, &buf);
// --- video ---
if (has_video) {
mxlGrainInfo video_grain{};
uint8_t* video_buf = nullptr;
mxlStatus vst = mxlFlowReaderGetGrainNonBlocking(
video_reader, video_index, &video_grain, &video_buf);
if (st == MXL_STATUS_OK) {
if (vst == MXL_STATUS_OK) {
frame_count++;
if (grain.flags & MXL_GRAIN_FLAG_INVALID) invalid_count++;
index++;
if (video_grain.flags & MXL_GRAIN_FLAG_INVALID) invalid_count++;
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);
std::memcpy(v210_frame.p_data, video_buf, video_stride * height);
NDIlib_util_V210_to_P216(&v210_frame, &p216_frame);
NDIlib_send_send_video_v2(ndi.sender, &p216_frame);
if (++ndi_frame_count == 1)
log("NDI receiver connected");
log("first NDI receiver connected");
} else {
ndi_frame_count = 0;
}
} else if (st == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) {
video_index++;
} else if (vst == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) {
mxlSleepForNs(1'000'000);
} else if (st == MXL_ERR_OUT_OF_RANGE_TOO_LATE) {
} else if (vst == MXL_ERR_OUT_OF_RANGE_TOO_LATE) {
late_count++;
mxlFlowRuntimeInfo ri{};
mxlFlowReaderGetRuntimeInfo(reader, &ri);
index = ri.headIndex;
mxlFlowReaderGetRuntimeInfo(video_reader, &ri);
video_index = ri.headIndex;
} else {
log("unexpected status=%d on index=%llu", st, index);
log("unexpected video status=%d on index=%llu", vst, video_index);
break;
}
@@ -136,11 +236,20 @@ class NDIOutNode : public dmf::NodeBase {
static_cast<double>(frame_count) / elapsed);
last_log_time = now;
}
} else if (!audio_advanced) {
// audio-only and nothing was ready — avoid busy spin
mxlSleepForNs(1'000'000);
}
}
if (has_video)
log("stopped — total frames=%llu invalid=%llu late=%llu",
frame_count, invalid_count, late_count);
mxlReleaseFlowReader(instance(), reader);
else
log("stopped");
if (has_video) mxlReleaseFlowReader(instance(), video_reader);
if (has_audio) mxlReleaseFlowReader(instance(), audio_reader);
}
};
nodes/testpattern/main.cpp
+38 -35
View File
@@ -21,18 +21,18 @@ class TestPatternNode : public dmf::NodeBase {
log("flow=%s %dx%d @ %d/%d fps pattern=%s",
flow_id.c_str(), width, height, fps_num, fps_den, pattern.c_str());
mxlFlowWriter writer{};
mxlFlowConfigInfo cfg_info{};
mxlFlowWriter video_writer{};
mxlFlowConfigInfo video_cfg{};
bool created = false;
mxlStatus st = mxlCreateFlowWriter(
mxlStatus vst = mxlCreateFlowWriter(
instance(),
dmf::make_video_flow_def(flow_id, node_id(), width, height, fps_num, fps_den).c_str(),
nullptr, &writer, &cfg_info, &created);
if (st != MXL_STATUS_OK) { log("video mxlCreateFlowWriter failed (status=%d)", st); return; }
nullptr, &video_writer, &video_cfg, &created);
if (vst != MXL_STATUS_OK) { log("video mxlCreateFlowWriter failed (status=%d)", vst); return; }
const uint32_t stride = cfg_info.discrete.sliceSizes[0];
const uint32_t video_stride = video_cfg.discrete.sliceSizes[0];
log("video stride=%u B/line grain=%u B ring=%u grains",
stride, stride * static_cast<uint32_t>(height), cfg_info.discrete.grainCount);
video_stride, video_stride * static_cast<uint32_t>(height), video_cfg.discrete.grainCount);
// --- audio flow (optional) ---
mxlFlowWriter audio_writer{};
@@ -42,16 +42,16 @@ class TestPatternNode : public dmf::NodeBase {
bool has_audio = config().contains("audio_flow_id");
if (has_audio) {
const auto audio_info = config().at("audio_flow_id");
const auto audio_id = audio_info.at("id").get<std::string>();
sample_rate = audio_info.value("sample_rate", 48000);
channels = audio_info.value("channels", 2);
const auto audio_flow_info = config().at("audio_flow_id");
const auto audio_flow_id = audio_flow_info.at("id").get<std::string>();
sample_rate = audio_flow_info.value("sample_rate", 48000);
channels = audio_flow_info.value("channels", 2);
log("audio flow=%s %d Hz %dch", audio_id.c_str(), sample_rate, channels);
log("audio flow=%s %d Hz %dch", audio_flow_id.c_str(), sample_rate, channels);
mxlStatus ast = mxlCreateFlowWriter(
instance(),
dmf::make_audio_flow_def(audio_id, node_id(), sample_rate, channels, 32,
dmf::make_audio_flow_def(audio_flow_id, node_id(), sample_rate, channels, 32,
fps_num, fps_den).c_str(),
nullptr, &audio_writer, &audio_cfg, &created);
if (ast != MXL_STATUS_OK) {
@@ -65,36 +65,39 @@ class TestPatternNode : public dmf::NodeBase {
// --- main loop ---
const mxlRational video_rate = {fps_num, fps_den};
const mxlRational audio_rate = {sample_rate, 1};
const size_t samples_per_frame =
static_cast<size_t>(sample_rate) * static_cast<size_t>(fps_den) / static_cast<size_t>(fps_num);
// -18 dBFS broadcast reference level
const float amplitude = static_cast<float>(std::pow(10.0, -18.0 / 20.0));
uint64_t index = mxlGetCurrentIndex(&video_rate);
uint64_t audio_index = has_audio ? mxlGetCurrentIndex(&audio_rate) : 0;
log("start video_index=%llu", index);
uint64_t video_index = mxlGetCurrentIndex(&video_rate);
uint64_t audio_index = 0;
if (has_audio) {
const mxlRational audio_rate = {sample_rate, 1};
audio_index = mxlGetCurrentIndex(&audio_rate);
}
log("start video_index=%llu", video_index);
while (dmf::g_running.load(std::memory_order_relaxed)) {
// --- video grain ---
mxlGrainInfo grain{};
uint8_t* buf = nullptr;
st = mxlFlowWriterOpenGrain(writer, index, &grain, &buf);
if (st != MXL_STATUS_OK) {
log("OpenGrain failed (status=%d), skipping index=%llu", st, index);
index++;
mxlGrainInfo video_grain{};
uint8_t* video_buf = nullptr;
vst = mxlFlowWriterOpenGrain(video_writer, video_index, &video_grain, &video_buf);
if (vst != MXL_STATUS_OK) {
log("OpenGrain failed (status=%d), skipping index=%llu", vst, video_index);
video_index++;
continue;
}
if (pattern == "bars") dmf::v210::fill_colorbars(buf, width, height, stride);
else if (pattern == "ire") dmf::v210::fill_ire_ramp (buf, width, height, stride);
else if (pattern == "black") dmf::v210::fill_black (buf, width, height, stride);
else if (pattern == "white") dmf::v210::fill_white (buf, width, height, stride);
else dmf::v210::fill_colorbars (buf, width, height, stride);
grain.flags = 0;
grain.validSlices = grain.totalSlices;
mxlFlowWriterCommitGrain(writer, &grain);
if (pattern == "bars") dmf::v210::fill_colorbars(video_buf, width, height, video_stride);
else if (pattern == "ire") dmf::v210::fill_ire_ramp (video_buf, width, height, video_stride);
else if (pattern == "black") dmf::v210::fill_black (video_buf, width, height, video_stride);
else if (pattern == "white") dmf::v210::fill_white (video_buf, width, height, video_stride);
else dmf::v210::fill_colorbars (video_buf, width, height, video_stride);
video_grain.flags = 0;
video_grain.validSlices = video_grain.totalSlices;
mxlFlowWriterCommitGrain(video_writer, &video_grain);
// --- audio: sine tones, channel c = (c+1) * 1000 Hz ---
if (has_audio) {
@@ -125,13 +128,13 @@ class TestPatternNode : public dmf::NodeBase {
audio_index += samples_per_frame;
}
const uint64_t ns = mxlGetNsUntilIndex(index + 1, &video_rate);
const uint64_t ns = mxlGetNsUntilIndex(video_index + 1, &video_rate);
if (ns > 0 && ns < 2'000'000'000ULL) mxlSleepForNs(ns);
index++;
video_index++;
}
log("stopped at video_index=%llu", index);
mxlReleaseFlowWriter(instance(), writer);
log("stopped at video_index=%llu", video_index);
mxlReleaseFlowWriter(instance(), video_writer);
if (has_audio) mxlReleaseFlowWriter(instance(), audio_writer);
}
};
studio-manager/main.cpp
+8 -6
View File
@@ -44,8 +44,8 @@ static dmf::FlowGraph build_graph() {
const std::string ndi_audio_flow = gen_uuid();
g.nodes = {
{ "testpattern", "testpattern", {{"pattern", "bars"}} },
{ "ndiin", "ndiin", {} },
{ "fakesink", "fakesink", {} },
// { "ndiin", "ndiin", {} },
// { "fakesink", "fakesink", {} },
{ "ndiout", "ndiout", {} },
};
const nlohmann::json video_fmt = {
@@ -55,10 +55,12 @@ static dmf::FlowGraph build_graph() {
{"kind","audio"}, {"sample_rate",48000}, {"channels",2}, {"bit_depth",32}
};
g.edges = {
{ tp_video_flow, "testpattern", "flow_id", "fakesink", "flow_id", video_fmt },
{ tp_audio_flow, "testpattern", "audio_flow_id", "", "", audio_fmt },
{ ndi_video_flow, "ndiin", "video_flow_id", "ndiout", "flow_id", video_fmt },
{ ndi_audio_flow, "ndiin", "audio_flow_id", "", "", audio_fmt },
// { tp_video_flow, "testpattern", "flow_id", "fakesink", "flow_id", video_fmt },
// { tp_audio_flow, "testpattern", "audio_flow_id", "", "", audio_fmt },
// { ndi_video_flow, "ndiin", "video_flow_id", "ndiout", "flow_id", video_fmt },
// { ndi_audio_flow, "ndiin", "audio_flow_id", "", "", audio_fmt },
{ tp_video_flow, "testpattern", "flow_id", "ndiout", "flow_id", video_fmt },
{ tp_audio_flow, "testpattern", "audio_flow_id", "ndiout", "audio_flow_id", audio_fmt },
};
return g;
}