Merge pull request 'Feature/ndi out node' (#1) from feature/ndi-out-node into main

Reviewed-on: #1

.vscode/c_cpp_properties.json

@@ -0,0 +1,17 @@
+{
+    "configurations": [
+        {
+            "name": "Linux",
+            "includePath": [
+                "${workspaceFolder}/**",
+                "${HOME}/SDK/NDI/include"
+            ],
+            "defines": [],
+            "compilerPath": "/usr/bin/clang",
+            "cStandard": "c17",
+            "cppStandard": "c++17",
+            "intelliSenseMode": "linux-clang-x64"
+        }
+    ],
+    "version": 4
+}

CMakeLists.txt

@@ -80,4 +80,11 @@ target_link_libraries(dmf-shared INTERFACE mxl nlohmann_json::nlohmann_json)
 
 add_subdirectory(nodes/testpattern)
 add_subdirectory(nodes/fakesink)
+
+set(NDI_SDK_DIR "" CACHE PATH "Path to NDI SDK root")
+if(NDI_SDK_DIR)
+    add_subdirectory(nodes/ndiout)
+    add_subdirectory(nodes/ndiin)
+endif()
+
 add_subdirectory(studio-manager)

nodes/ndiin/CMakeLists.txt

@@ -0,0 +1,11 @@
+add_executable(dmf-node-ndiin main.cpp)
+target_compile_features(dmf-node-ndiin PRIVATE cxx_std_20)
+target_link_libraries(dmf-node-ndiin PRIVATE dmf-shared)
+install(TARGETS dmf-node-ndiin RUNTIME DESTINATION bin)
+
+set(NDI_INCLUDE "${NDI_SDK_DIR}/include")
+target_include_directories(dmf-node-ndiin PRIVATE
+        "${NDI_INCLUDE}"
+)
+find_library(NDI_LIB NAMES ndi PATHS "${NDI_SDK_DIR}/lib/${CMAKE_SYSTEM_PROCESSOR}-linux-gnu" "${NDI_SDK_DIR}/lib/x64" "${NDI_SDK_DIR}/lib" NO_DEFAULT_PATH)
+target_link_libraries(dmf-node-ndiin PRIVATE ${NDI_LIB})

nodes/ndiin/main.cpp

@@ -0,0 +1,161 @@
+#include <cstring>
+#include <string>
+#include <vector>
+#include <mxl/flow.h>
+#include <mxl/time.h>
+#include "NodeBase.hpp"
+#include "FlowDef.hpp"
+#include "NDIReceiver.hpp"
+
+class NDIInNode : public dmf::NodeBase {
+    void run() override {
+        const auto source_num = static_cast<uint32_t>(config().value("source_num", 0));
+
+        dmf::NDIReceiver ndi;
+        dmf::NDIReceiver::SourceInfo src;
+        try {
+            auto sources = ndi.find_sources(5000);
+            log("Available NDI sources:");
+            for (const auto& name : sources)
+                log("  %s", name.c_str());
+            ndi.connect(source_num);
+            src = ndi.probe();
+        } catch (const std::runtime_error& e) {
+            log("Error: %s", e.what());
+            return;
+        }
+
+        // --- video flow ---
+        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",   src.width);
+        const int  height          = video_flow_info.value("height",  src.height);
+        const int  fps_num         = video_flow_info.value("fps_num", src.fps_num);
+        const int  fps_den         = video_flow_info.value("fps_den", src.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{};
+        mxlFlowConfigInfo video_cfg{};
+        bool created = false;
+        mxlStatus st = mxlCreateFlowWriter(
+            instance(),
+            dmf::make_video_flow_def(video_flow_id, node_id(), width, height, fps_num, fps_den).c_str(),
+            nullptr, &video_writer, &video_cfg, &created);
+        if (st != MXL_STATUS_OK) { log("video mxlCreateFlowWriter failed (status=%d)", st); 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);
+
+        // --- audio flow (optional — only created when graph wires audio_flow_id) ---
+        mxlFlowWriter audio_writer{};
+        mxlFlowConfigInfo audio_cfg{};
+        int  sample_rate  = 0;
+        int  channels     = 0;
+        int  bit_depth    = 32;
+        bool has_audio    = config().contains("audio_flow_id");
+
+        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);
+            bit_depth   = audio_flow_info.value("bit_depth", 32);
+
+            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(),
+                nullptr, &audio_writer, &audio_cfg, &created);
+            if (ast != MXL_STATUS_OK) {
+                log("audio mxlCreateFlowWriter failed (status=%d) — continuing without audio", ast);
+                has_audio = false;
+            } else {
+                log("audio channels=%u  buffer=%u samples",
+                    audio_cfg.continuous.channelCount, audio_cfg.continuous.bufferLength);
+            }
+        }
+
+        // --- 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;
+        log("start video_index=%llu", video_index);
+
+        std::vector<uint8_t> latest_video(video_stride * height);
+        bool have_video = false;
+        std::vector<float> audio_buf;
+        dmf::NDIReceiver::AudioInfo audio_info;
+
+        while (dmf::g_running.load(std::memory_order_relaxed)) {
+            dmf::NDIReceiver::FrameKind kind;
+            try {
+                kind = ndi.capture(latest_video.data(), video_stride, audio_buf, audio_info);
+            } catch (const std::runtime_error& e) {
+                log("NDI error: %s — stopping", e.what());
+                break;
+            }
+
+            if (kind == dmf::NDIReceiver::FrameKind::Video) {
+                have_video = true;
+            } else if (kind == dmf::NDIReceiver::FrameKind::Audio && has_audio) {
+                mxlMutableWrappedMultiBufferSlice slice{};
+                if (mxlFlowWriterOpenSamples(audio_writer, audio_index,
+                                             static_cast<size_t>(audio_info.samples), &slice) == MXL_STATUS_OK) {
+                    // MXL audio is float32 planar: each channel occupies its own ring buffer
+                    // region, accessed at base + c * stride. Fragments handle ring wraparound.
+                    const size_t frag0 = slice.base.fragments[0].size / sizeof(float);
+                    const size_t frag1 = slice.base.fragments[1].size / sizeof(float);
+                    for (int c = 0; c < audio_info.channels; ++c) {
+                        const float* src = &audio_buf[c * audio_info.channel_stride];
+                        auto* dst0 = reinterpret_cast<float*>(
+                            static_cast<uint8_t*>(slice.base.fragments[0].pointer) + c * slice.stride);
+                        std::memcpy(dst0, src, frag0 * sizeof(float));
+                        if (frag1 > 0) {
+                            auto* dst1 = reinterpret_cast<float*>(
+                                static_cast<uint8_t*>(slice.base.fragments[1].pointer) + c * slice.stride);
+                            std::memcpy(dst1, src + frag0, frag1 * sizeof(float));
+                        }
+                    }
+                    mxlFlowWriterCommitSamples(audio_writer);
+                }
+                audio_index += audio_info.samples;
+            }
+
+            // Write video grain whenever the MXL clock has reached video_index
+            const uint64_t current = mxlGetCurrentIndex(&video_rate);
+            if (current >= video_index) {
+                mxlGrainInfo grain{};
+                uint8_t* buf = nullptr;
+                st = mxlFlowWriterOpenGrain(video_writer, video_index, &grain, &buf);
+                if (st == MXL_STATUS_OK) {
+                    if (have_video) {
+                        std::memcpy(buf, latest_video.data(), latest_video.size());
+                        grain.flags = 0;
+                    } else {
+                        grain.flags = MXL_GRAIN_FLAG_INVALID;
+                    }
+                    grain.validSlices = grain.totalSlices;
+                    mxlFlowWriterCommitGrain(video_writer, &grain);
+                } else {
+                    log("video OpenGrain failed (status=%d) at index=%llu", st, video_index);
+                }
+                video_index = current + 1;
+            }
+        }
+
+        log("stopped at video_index=%llu", video_index);
+        mxlReleaseFlowWriter(instance(), video_writer);
+        if (has_audio) mxlReleaseFlowWriter(instance(), audio_writer);
+    }
+};
+
+int main() {
+    NDIInNode node;
+    return node.execute();
+}

nodes/ndiout/CMakeLists.txt

@@ -0,0 +1,11 @@
+add_executable(dmf-node-ndiout main.cpp)
+target_compile_features(dmf-node-ndiout PRIVATE cxx_std_20)
+target_link_libraries(dmf-node-ndiout PRIVATE dmf-shared)
+install(TARGETS dmf-node-ndiout RUNTIME DESTINATION bin)
+
+set(NDI_INCLUDE "${NDI_SDK_DIR}/include")
+target_include_directories(dmf-node-ndiout PRIVATE
+        "${NDI_INCLUDE}"
+)
+find_library(NDI_LIB NAMES ndi PATHS "${NDI_SDK_DIR}/lib/${CMAKE_SYSTEM_PROCESSOR}-linux-gnu" "${NDI_SDK_DIR}/lib/x64" "${NDI_SDK_DIR}/lib" NO_DEFAULT_PATH)
+target_link_libraries(dmf-node-ndiout PRIVATE ${NDI_LIB})

nodes/ndiout/main.cpp

@@ -0,0 +1,150 @@
+#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();
+}

nodes/testpattern/main.cpp

@@ -1,3 +1,5 @@
+#include <cmath>
+#include <numbers>
 #include <string>
 #include <mxl/flow.h>
 #include <mxl/time.h>
@@ -7,41 +9,77 @@
 
 class TestPatternNode : public dmf::NodeBase {
     void run() override {
+        // --- video flow ---
         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);
+        const auto pattern   = config().value("pattern", "bars");
 
-        log("flow=%s  %dx%d @ %d/%d fps", flow_id.c_str(), width, height, fps_num, fps_den);
+        log("flow=%s  %dx%d @ %d/%d fps  pattern=%s",
-
+            flow_id.c_str(), width, height, fps_num, fps_den, pattern.c_str());
-        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);
+            instance(),
-        if (st != MXL_STATUS_OK) {
+            dmf::make_video_flow_def(flow_id, node_id(), width, height, fps_num, fps_den).c_str(),
-            log("mxlCreateFlowWriter failed (status=%d)", st);
+            nullptr, &writer, &cfg_info, &created);
-            return;
+        if (st != MXL_STATUS_OK) { log("video 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",
+        log("video stride=%u B/line  grain=%u B  ring=%u grains",
             stride, stride * static_cast<uint32_t>(height), cfg_info.discrete.grainCount);
 
-        const mxlRational rate  = {fps_num, fps_den};
+        // --- audio flow (optional) ---
-        uint64_t          index = mxlGetCurrentIndex(&rate);
+        mxlFlowWriter     audio_writer{};
-        log("start index=%llu", index);
+        mxlFlowConfigInfo audio_cfg{};
+        int  sample_rate      = 48000;
+        int  channels         = 2;
+        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);
+
+            log("audio flow=%s  %d Hz  %dch", audio_id.c_str(), sample_rate, channels);
+
+            mxlStatus ast = mxlCreateFlowWriter(
+                instance(),
+                dmf::make_audio_flow_def(audio_id, node_id(), sample_rate, channels, 32,
+                                         fps_num, fps_den).c_str(),
+                nullptr, &audio_writer, &audio_cfg, &created);
+            if (ast != MXL_STATUS_OK) {
+                log("audio mxlCreateFlowWriter failed (status=%d) — continuing without audio", ast);
+                has_audio = false;
+            } else {
+                log("audio channels=%u  buffer=%u samples",
+                    audio_cfg.continuous.channelCount, audio_cfg.continuous.bufferLength);
+            }
+        }
+
+        // --- 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);
 
         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);
@@ -49,18 +87,52 @@ class TestPatternNode : public dmf::NodeBase {
                 continue;
             }
 
-            dmf::v210::fill_frame(buf, width, height, stride);
+            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;  // mark grain complete so readers can consume it
+            grain.validSlices = grain.totalSlices;
             mxlFlowWriterCommitGrain(writer, &grain);
 
-            const uint64_t ns = mxlGetNsUntilIndex(index + 1, &rate);
+            // --- audio: sine tones, channel c = (c+1) * 1000 Hz ---
+            if (has_audio) {
+                mxlMutableWrappedMultiBufferSlice slice{};
+                if (mxlFlowWriterOpenSamples(audio_writer, audio_index,
+                                             samples_per_frame, &slice) == MXL_STATUS_OK) {
+                    const size_t frag0 = slice.base.fragments[0].size / sizeof(float);
+                    const size_t frag1 = slice.base.fragments[1].size / sizeof(float);
+                    for (int c = 0; c < channels; ++c) {
+                        const double freq   = 1000.0 * (c + 1);
+                        const double period = static_cast<double>(sample_rate) / freq;
+                        auto write_samples = [&](float* dst, size_t count, size_t offset) {
+                            for (size_t s = 0; s < count; ++s)
+                                dst[s] = amplitude * static_cast<float>(
+                                    std::sin(2.0 * std::numbers::pi * (audio_index + offset + s) / period));
+                        };
+                        auto* dst0 = reinterpret_cast<float*>(
+                            static_cast<uint8_t*>(slice.base.fragments[0].pointer) + c * slice.stride);
+                        write_samples(dst0, frag0, 0);
+                        if (frag1 > 0) {
+                            auto* dst1 = reinterpret_cast<float*>(
+                                static_cast<uint8_t*>(slice.base.fragments[1].pointer) + c * slice.stride);
+                            write_samples(dst1, frag1, frag0);
+                        }
+                    }
+                    mxlFlowWriterCommitSamples(audio_writer);
+                }
+                audio_index += samples_per_frame;
+            }
+
+            const uint64_t ns = mxlGetNsUntilIndex(index + 1, &video_rate);
             if (ns > 0 && ns < 2'000'000'000ULL) mxlSleepForNs(ns);
             index++;
         }
 
-        log("stopped at index=%llu", index);
+        log("stopped at video_index=%llu", index);
         mxlReleaseFlowWriter(instance(), writer);
+        if (has_audio) mxlReleaseFlowWriter(instance(), audio_writer);
     }
 };
 

shared/FlowDef.hpp

@@ -36,4 +36,40 @@ inline std::string make_video_flow_def(
     }.dump();
 }
 
+// Generates a minimal but valid NMOS IS-04 flow definition JSON string
+// for a raw PCM audio flow. grain_rate_num/den sets the grain delivery rate
+// (default 25/1 = one grain per video frame). MXL computes grain size as
+// sample_rate / grain_rate samples per grain.
+inline std::string make_audio_flow_def(
+    const std::string& flow_id,
+    const std::string& label,
+    int sample_rate,
+    int channels,
+    int bit_depth       = 32,
+    int grain_rate_num  = 25,
+    int grain_rate_den  = 1)
+{
+    using json = nlohmann::json;
+    static const char* ch_labels[] = {"L","R","C","LFE","Ls","Rs","Lss","Rss"};
+    json ch_arr = json::array();
+    for (int i = 0; i < channels; ++i)
+        ch_arr.push_back({{"label", i < 8 ? ch_labels[i] : ("Ch" + std::to_string(i + 1))}});
+    return json{
+        {"id",          flow_id},
+        {"format",      "urn:x-nmos:format:audio"},
+        {"label",       label},
+        {"description", label + " MXL Audio Flow"},
+        {"media_type",  "audio/L" + std::to_string(bit_depth)},
+        {"parents",     json::array()},
+        {"grain_rate",  {{"numerator", grain_rate_num}, {"denominator", grain_rate_den}}},
+        {"sample_rate", {{"numerator", sample_rate},    {"denominator", 1}}},
+        {"channels",       ch_arr},   // NMOS IS-04 metadata (ignored by MXL)
+        {"channel_count",  channels},  // MXL uses this for grain buffer geometry
+        {"bit_depth",      bit_depth}, // must be 32 or 64
+        {"tags", {
+            {"urn:x-nmos:tag:grouphint/v1.0", json::array({label + ":Audio"})}
+        }},
+    }.dump();
+}
+
 } // namespace dmf

shared/NDIReceiver.hpp

@@ -0,0 +1,231 @@
+#pragma once
+
+#include <stdexcept>
+#include <string>
+#include <vector>
+#include <Processing.NDI.Lib.h>
+#include "Signal.hpp"
+#include "V210.hpp"
+
+namespace dmf {
+
+class NDIReceiver {
+public:
+    struct SourceInfo {
+        int width   = 0;
+        int height  = 0;
+        int fps_num = 0;
+        int fps_den = 0;
+        int stride  = 0;
+        NDIlib_FourCC_type_e fourcc{};
+    };
+
+    NDIReceiver() {
+        if (!NDIlib_is_supported_CPU())
+            throw std::runtime_error("CPU is not sufficient for NDI");
+        if (!NDIlib_initialize())
+            throw std::runtime_error("NDI lib init failed");
+    }
+
+    ~NDIReceiver() {
+        if (recv_) NDIlib_recv_destroy(recv_);
+        if (find_) NDIlib_find_destroy(find_);
+        NDIlib_destroy();
+    }
+
+    // Discovers NDI sources on the network. Polls up to max_attempts times,
+    // each waiting timeout_ms milliseconds. Respects g_running.
+    std::vector<std::string> find_sources(uint32_t timeout_ms, int max_attempts = 10) {
+        find_ = NDIlib_find_create_v2();
+        if (!find_)
+            throw std::runtime_error("Cannot create NDI finder");
+
+        uint32_t count = 0;
+        const NDIlib_source_t* p_sources = nullptr;
+        for (int i = 0; !count && i < max_attempts; ++i) {
+            if (!dmf::g_running.load(std::memory_order_relaxed)) {
+                NDIlib_find_destroy(find_);
+                find_ = nullptr;
+                throw std::runtime_error("Interrupted while searching for NDI sources");
+            }
+            NDIlib_find_wait_for_sources(find_, timeout_ms);
+            p_sources = NDIlib_find_get_current_sources(find_, &count);
+        }
+        if (!count) {
+            NDIlib_find_destroy(find_);
+            find_ = nullptr;
+            throw std::runtime_error("No NDI sources found after timeout");
+        }
+
+        // Copy while finder is alive: p_sources points into finder-owned memory
+        std::vector<std::string> names;
+        for (uint32_t i = 0; i < count; ++i) {
+            names.emplace_back(p_sources[i].p_ndi_name);
+            source_names_.emplace_back(p_sources[i].p_ndi_name);
+            source_urls_.emplace_back(p_sources[i].p_url_address);
+        }
+        sources_.reserve(source_names_.size());
+        for (size_t i = 0; i < source_names_.size(); ++i)
+            sources_.push_back({source_names_[i].c_str(), source_urls_[i].c_str()});
+
+        NDIlib_find_destroy(find_);
+        find_ = nullptr;
+        return names;
+    }
+
+    // Connects to a discovered source by index.
+    void connect(uint32_t source_num) {
+        if (sources_.empty())
+            throw std::runtime_error("No sources available — call find_sources() first");
+        if (source_num >= static_cast<uint32_t>(sources_.size()))
+            throw std::runtime_error("source_num exceeds available source count");
+
+        recv_ = NDIlib_recv_create_v3();
+        if (!recv_)
+            throw std::runtime_error("Cannot create NDI receive instance");
+        NDIlib_recv_connect(recv_, &sources_[source_num]);
+    }
+
+    // Captures the first video frame to determine resolution, frame rate, and format.
+    // Stores the result internally for use by capture_v210(). Respects g_running.
+    SourceInfo probe() {
+        while (dmf::g_running.load(std::memory_order_relaxed)) {
+            NDIlib_video_frame_v2_t frame;
+            auto type = NDIlib_recv_capture_v3(recv_, &frame, nullptr, nullptr, 1000);
+            if (type == NDIlib_frame_type_error)
+                throw std::runtime_error("NDI source lost during probe");
+            if (type != NDIlib_frame_type_video)
+                continue;
+
+            info_.width   = frame.xres;
+            info_.height  = frame.yres;
+            info_.fps_num = frame.frame_rate_N;
+            info_.fps_den = frame.frame_rate_D;
+            info_.fourcc  = static_cast<NDIlib_FourCC_type_e>(frame.FourCC);
+            info_.stride  = frame.line_stride_in_bytes;
+            if (info_.stride == 0)
+                info_.stride = info_.width * bytes_per_pixel(info_.fourcc);
+
+            NDIlib_recv_free_video_v2(recv_, &frame);
+            return info_;
+        }
+        throw std::runtime_error("Interrupted during probe");
+    }
+
+    enum class FrameKind { None, Video, Audio };
+
+    struct AudioInfo {
+        int sample_rate    = 0;
+        int channels       = 0;
+        int samples        = 0;
+        int channel_stride = 0;  // floats between channel planes (NDI planar layout)
+    };
+
+    // Receives one NDI frame. On video: converts to V210 in frame_buffer/frame_stride.
+    // On audio: copies float32 planar samples into audio_out and fills audio_info.
+    // Returns FrameKind::None on timeout or non-A/V frames.
+    // Throws on source lost or video format change.
+    FrameKind capture(uint8_t* frame_buffer, uint32_t frame_stride,
+                      std::vector<float>& audio_out, AudioInfo& audio_info) {
+        NDIlib_video_frame_v2_t video_frame{};
+        NDIlib_audio_frame_v3_t audio_frame{};
+        auto type = NDIlib_recv_capture_v3(recv_, &video_frame, &audio_frame, nullptr, 5);
+
+        if (type == NDIlib_frame_type_error)
+            throw std::runtime_error("NDI source lost");
+
+        if (type == NDIlib_frame_type_status_change) {
+            const SourceInfo old = info_;
+            info_ = probe();
+            if (info_.width   != old.width  || info_.height  != old.height ||
+                info_.fps_num != old.fps_num || info_.fps_den != old.fps_den) {
+                throw std::runtime_error(
+                    "source format changed: " +
+                    std::to_string(old.width) + "x" + std::to_string(old.height) +
+                    " @" + std::to_string(old.fps_num) + "/" + std::to_string(old.fps_den) +
+                    " -> " +
+                    std::to_string(info_.width) + "x" + std::to_string(info_.height) +
+                    " @" + std::to_string(info_.fps_num) + "/" + std::to_string(info_.fps_den));
+            }
+            return FrameKind::None;
+        }
+
+        if (type == NDIlib_frame_type_video) {
+            switch (info_.fourcc) {
+                case NDIlib_FourCC_type_UYVY:
+                    v210::UYVYtoV210(video_frame.p_data, frame_buffer,
+                                     info_.width, info_.height, info_.stride, frame_stride);
+                    break;
+                case NDIlib_FourCC_type_P216: {
+                    NDIlib_video_frame_v2_t dst{};
+                    dst.p_data = frame_buffer;
+                    dst.line_stride_in_bytes = frame_stride;
+                    NDIlib_util_P216_to_V210(&video_frame, &dst);
+                    break;
+                }
+                default:
+                    NDIlib_recv_free_video_v2(recv_, &video_frame);
+                    throw std::runtime_error("Unsupported NDI color format: " + fourcc_str(info_.fourcc));
+            }
+            NDIlib_recv_free_video_v2(recv_, &video_frame);
+            return FrameKind::Video;
+        }
+
+        if (type == NDIlib_frame_type_audio) {
+            if (audio_frame.FourCC != NDIlib_FourCC_audio_type_FLTP) {
+                NDIlib_recv_free_audio_v3(recv_, &audio_frame);
+                return FrameKind::None;  // compressed or unknown format — skip
+            }
+            audio_info.sample_rate    = audio_frame.sample_rate;
+            audio_info.channels       = audio_frame.no_channels;
+            audio_info.samples        = audio_frame.no_samples;
+            audio_info.channel_stride = audio_frame.channel_stride_in_bytes / sizeof(float);
+            const int total = audio_frame.no_channels * audio_info.channel_stride;
+            const auto* fdata = reinterpret_cast<const float*>(audio_frame.p_data);
+            audio_out.assign(fdata, fdata + total);
+            NDIlib_recv_free_audio_v3(recv_, &audio_frame);
+            return FrameKind::Audio;
+        }
+
+        return FrameKind::None;
+    }
+
+private:
+    NDIlib_find_instance_t       find_ = nullptr;
+    NDIlib_recv_instance_t       recv_ = nullptr;
+    SourceInfo                   info_;
+    std::vector<std::string>     source_names_;
+    std::vector<std::string>     source_urls_;
+    std::vector<NDIlib_source_t> sources_;
+
+    static int bytes_per_pixel(NDIlib_FourCC_type_e fc) {
+        switch (fc) {
+            case NDIlib_FourCC_video_type_UYVY:
+            case NDIlib_FourCC_video_type_YV12:
+            case NDIlib_FourCC_video_type_I420:
+            case NDIlib_FourCC_video_type_NV12:  return 2;
+            case NDIlib_FourCC_video_type_BGRA:
+            case NDIlib_FourCC_video_type_RGBA:
+            case NDIlib_FourCC_video_type_BGRX:
+            case NDIlib_FourCC_video_type_RGBX:  return 4;
+            case NDIlib_FourCC_video_type_UYVA:  return 3;
+            case NDIlib_FourCC_video_type_P216:  return 4;
+            case NDIlib_FourCC_video_type_PA16:  return 6;
+            default:                              return 2;
+        }
+    }
+
+    static std::string fourcc_str(NDIlib_FourCC_type_e fc) {
+        uint32_t v = static_cast<uint32_t>(fc);
+        char s[5] = {
+            static_cast<char>((v >>  0) & 0xFF),
+            static_cast<char>((v >>  8) & 0xFF),
+            static_cast<char>((v >> 16) & 0xFF),
+            static_cast<char>((v >> 24) & 0xFF),
+            '\0'
+        };
+        return s;
+    }
+};
+
+} // namespace dmf

shared/V210.hpp

@@ -2,6 +2,7 @@
 #include <array>
 #include <cstddef>
 #include <cstdint>
+#include <cstring>
 
 namespace dmf::v210 {
 
@@ -82,11 +83,90 @@ inline void write_bar_line(uint8_t* line, int width, uint32_t /*stride*/)
     }
 }
 
-// Fill an entire frame buffer with color bars.
+// Write one horizontal line of an arbitrary bar palette.
+template<std::size_t N>
+inline void write_palette_line(uint8_t* line, int width, const std::array<Color, N>& palette)
+{
+    const int n = static_cast<int>(N);
+    const int blocks = width / 6;
+    for (int b = 0; b < blocks; b++) {
+        int x = b * 6;
+        auto color = [&](int px) -> const Color& {
+            return palette[static_cast<std::size_t>(px * n / width)];
+        };
+        const Color& c01 = color(x);
+        const Color& c23 = color(x + 2);
+        const Color& c45 = color(x + 4);
+        pack_block(line + b * 16,
+            c01, c01.y, color(x+1).y,
+            c23, c23.y, color(x+3).y,
+            c45, c45.y, color(x+5).y);
+    }
+}
+
+// Fill a frame with a solid color.
+inline void fill_solid(uint8_t* buf, int width, int height, uint32_t stride, Color c)
+{
+    const int blocks = width / 6;
+    for (int b = 0; b < blocks; b++)
+        pack_block(buf + b * 16, c, c.y, c.y, c, c.y, c.y, c, c.y, c.y);
+    for (int y = 1; y < height; y++)
+        std::memcpy(buf + static_cast<ptrdiff_t>(y) * stride, buf, blocks * 16);
+}
+
+// SMPTE 75% color bars.
+inline void fill_colorbars(uint8_t* buf, int width, int height, uint32_t stride)
+{
+    for (int y = 0; y < height; y++)
+        write_palette_line(buf + static_cast<ptrdiff_t>(y) * stride, width, SMPTE_BARS);
+}
+
+// IRE 11-step greyscale ramp.
+inline void fill_ire_ramp(uint8_t* buf, int width, int height, uint32_t stride)
+{
+    for (int y = 0; y < height; y++)
+        write_palette_line(buf + static_cast<ptrdiff_t>(y) * stride, width, IRE_BARS);
+}
+
+// 10-bit limited black  (Y=64,  Cb=Cr=512).
+inline void fill_black(uint8_t* buf, int width, int height, uint32_t stride)
+{
+    fill_solid(buf, width, height, stride, {64, 512, 512});
+}
+
+// 10-bit limited white  (Y=940, Cb=Cr=512).
+inline void fill_white(uint8_t* buf, int width, int height, uint32_t stride)
+{
+    fill_solid(buf, width, height, stride, {940, 512, 512});
+}
+
+// Kept for backward compatibility.
 inline void fill_frame(uint8_t* buf, int width, int height, uint32_t stride)
 {
+    fill_colorbars(buf, width, height, stride);
+}
+
+inline void UYVYtoV210(uint8_t* src_buf, uint8_t* dst_buf, int width, int height, uint32_t src_stride, uint32_t dst_stride)
+{
+    const uint8_t* src = src_buf;
+    uint8_t* dst = dst_buf;
+    const int blocks = width / 6;
+
     for (int y = 0; y < height; y++) {
-        write_bar_line(buf + static_cast<ptrdiff_t>(y) * stride, width, stride);
+        for (int b = 0; b < blocks; b++) {
+            const uint8_t* mp = src + b * 12;  // 3 macropixels = 12 bytes
+            // mp[0]=U0, mp[1]=Y0, mp[2]=V0, mp[3]=Y1
+            // mp[4]=U1, mp[5]=Y2, mp[6]=V1, mp[7]=Y3
+            // mp[8]=U2, mp[9]=Y4, mp[10]=V2, mp[11]=Y5
+
+            dmf::v210::pack_block(dst + b * 16,
+                {0, (uint16_t)(mp[0]<<2), (uint16_t)(mp[2]<<2)}, (uint16_t)(mp[1]<<2), (uint16_t)(mp[3]<<2),
+                {0, (uint16_t)(mp[4]<<2), (uint16_t)(mp[6]<<2)}, (uint16_t)(mp[5]<<2), (uint16_t)(mp[7]<<2),
+                {0, (uint16_t)(mp[8]<<2), (uint16_t)(mp[10]<<2)}, (uint16_t)(mp[9]<<2), (uint16_t)(mp[11]<<2)
+            );
+        }
+        src += src_stride;
+        dst += dst_stride;
     }
 }
 

studio-manager/main.cpp

@@ -38,13 +38,27 @@ static std::string gen_uuid() {
 
 static dmf::FlowGraph build_graph() {
     dmf::FlowGraph g;
+    const std::string tp_video_flow = gen_uuid();
+    const std::string tp_audio_flow = gen_uuid();
+    const std::string ndi_video_flow = gen_uuid();
+    const std::string ndi_audio_flow = gen_uuid();
     g.nodes = {
-        { "testpattern", "testpattern", {} },
+        { "testpattern", "testpattern", {{"pattern", "bars"}} },
+        { "ndiin",       "ndiin",       {} },
         { "fakesink",    "fakesink",    {} },
+        { "ndiout",      "ndiout",      {} },
+    };
+    const nlohmann::json video_fmt = {
+        {"kind","video"}, {"width",1920}, {"height",1080}, {"fps_num",25}, {"fps_den",1}
+    };
+    const nlohmann::json audio_fmt = {
+        {"kind","audio"}, {"sample_rate",48000}, {"channels",2}, {"bit_depth",32}
     };
     g.edges = {
-        { gen_uuid(), "testpattern", "flow_id", "fakesink", "flow_id",
+        { tp_video_flow,  "testpattern", "flow_id",       "fakesink", "flow_id", video_fmt },
-          { {"kind","video"}, {"width",1920}, {"height",1080}, {"fps_num",25}, {"fps_den",1} } },
+        { 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 },
     };
     return g;
 }