Read info about source from source/flow_def

CMakeLists.txt

@@ -77,7 +77,7 @@ FetchContent_MakeAvailable(json)
 find_package(PkgConfig REQUIRED)
 
 # Check for FFmpeg components
-pkg_check_modules(FFMPEG REQUIRED
+pkg_check_modules(FFMPEG REQUIRED IMPORTED_TARGET
     libavformat
     libavcodec
     libswscale
@@ -85,10 +85,9 @@ pkg_check_modules(FFMPEG REQUIRED
     libswresample
 )
 
-# Create an interface library for FFmpeg
+# Wrap in a named alias so nodes just link against "ffmpeg"
 add_library(ffmpeg INTERFACE)
-target_include_directories(ffmpeg INTERFACE ${FFMPEG_INCLUDE_DIRS})
+target_link_libraries(ffmpeg INTERFACE PkgConfig::FFMPEG)
-target_link_libraries(ffmpeg INTERFACE ${FFMPEG_LIBRARIES})
 
 # ── Shared utilities (Signal.hpp, NodeBase.hpp, FlowDef.hpp, V210.hpp) ───────
 add_library(dmf-shared INTERFACE)

nodes/decklinkin/main.cpp

@@ -37,10 +37,10 @@ class DeckLinkInNode : public dmf::NodeBase {
         if (!config().contains("video_flow_id")) { log("no video output connected — exiting"); return; }
         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",   vi.width);
+        const int   width   = vi.width;
-        const int   height  = video_flow_info.value("height",  vi.height);
+        const int   height  = vi.height;
-        const int   fps_num = video_flow_info.value("fps_num", vi.fps_num);
+        const int   fps_num = vi.fps_num;
-        const int   fps_den = video_flow_info.value("fps_den", vi.fps_den);
+        const int   fps_den = vi.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 = nullptr;

nodes/decklinkout/main.cpp

@@ -5,6 +5,7 @@
 #include <mxl/flow.h>
 #include <mxl/time.h>
 #include "NodeBase.hpp"
+#include "FlowDef.hpp"
 #include "DeckLinkSender.hpp"
 
 class DeckLinkOutNode : public dmf::NodeBase {
@@ -23,13 +24,7 @@ class DeckLinkOutNode : public dmf::NodeBase {
         uint32_t      video_stride = 0;
 
         if (has_video) {
-            const auto flow_info = config().at("video_flow_id");
+            flow_id = config().at("video_flow_id").at("id").get<std::string>();
-            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;
@@ -38,7 +33,6 @@ class DeckLinkOutNode : public dmf::NodeBase {
                 if (!active) mxlSleepForNs(100'000'000);
             }
             if (!dmf::g_running) return;
-            log("flow active — starting read");
 
             mxlFlowConfigInfo video_cfg{};
             mxlStatus vst = mxlCreateFlowReader(instance(), flow_id.c_str(), "", &video_reader);
@@ -48,6 +42,15 @@ class DeckLinkOutNode : public dmf::NodeBase {
             }
             mxlFlowReaderGetConfigInfo(video_reader, &video_cfg);
             video_stride = video_cfg.discrete.sliceSizes[0];
+
+            // Read actual format from what the source wrote — never trust config values here
+            const auto vfi = dmf::read_video_flow_info(domain(), flow_id);
+            width   = vfi.width;
+            height  = vfi.height;
+            fps_num = vfi.fps_num;
+            fps_den = vfi.fps_den;
+            log("video flow=%s  %dx%d @ %d/%d fps  stride=%u",
+                flow_id.c_str(), width, height, fps_num, fps_den, video_stride);
         }
 
         // --- audio flow (optional) ---
@@ -59,14 +62,18 @@ class DeckLinkOutNode : public dmf::NodeBase {
         std::string audio_flow_id;
 
         if (has_audio) {
-            const auto audio_flow_info = config().at("audio_flow_id");
+            audio_flow_id = config().at("audio_flow_id").at("id").get<std::string>();
-            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;
-            log("audio flow=%s  %d Hz  %dch  %d samples/frame",
-                audio_flow_id.c_str(), sample_rate, channels, samples_per_frame);
 
+            log("waiting for audio flow to become active...");
+            bool active = false;
+            while (!active && dmf::g_running.load(std::memory_order_relaxed)) {
+                mxlIsFlowActive(instance(), audio_flow_id.c_str(), &active);
+                if (!active) mxlSleepForNs(100'000'000);
+            }
+            if (!dmf::g_running) { has_audio = false; }
+        }
+
+        if (has_audio) {
             mxlStatus ast = mxlCreateFlowReader(instance(), audio_flow_id.c_str(), "", &audio_reader);
             if (ast != MXL_STATUS_OK) {
                 log("audio mxlCreateFlowReader failed (%s) — continuing without audio",
@@ -75,8 +82,14 @@ class DeckLinkOutNode : public dmf::NodeBase {
             } else {
                 mxlFlowConfigInfo audio_cfg{};
                 mxlFlowReaderGetConfigInfo(audio_reader, &audio_cfg);
-                log("audio channels=%u  buffer=%u samples",
+                const auto afi = dmf::read_audio_flow_info(domain(), audio_flow_id);
-                    audio_cfg.continuous.channelCount, audio_cfg.continuous.bufferLength);
+                sample_rate       = afi.sample_rate;
+                channels          = afi.channels;
+                samples_per_frame = afi.samples_per_grain;
+                log("audio flow=%s  %d Hz  %dch  %d samples/frame",
+                    audio_flow_id.c_str(), sample_rate, channels, samples_per_frame);
+                log("audio mxl buffer=%u samples",
+                    audio_cfg.continuous.bufferLength);
             }
         }
 

nodes/ndiin/main.cpp

@@ -30,10 +30,10 @@ class NDIInNode : public dmf::NodeBase {
         if (!config().contains("video_flow_id")) { log("no video output connected — exiting"); return; }
         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  width           = src.width;
-        const int  height          = video_flow_info.value("height",  src.height);
+        const int  height          = src.height;
-        const int  fps_num         = video_flow_info.value("fps_num", src.fps_num);
+        const int  fps_num         = src.fps_num;
-        const int  fps_den         = video_flow_info.value("fps_den", src.fps_den);
+        const int  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);
 
@@ -61,9 +61,19 @@ class NDIInNode : public dmf::NodeBase {
         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>();
+            bit_depth = audio_flow_info.value("bit_depth", 32);
+
+            log("probing audio format from NDI source...");
+            try {
+                const auto ap = ndi.probe_audio();
+                sample_rate = ap.sample_rate;
+                channels    = ap.channels;
+                log("detected audio: %d Hz  %dch", sample_rate, channels);
+            } catch (const std::runtime_error& e) {
+                log("audio probe failed (%s) — using config defaults", e.what());
                 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);
 

nodes/ndiout/main.cpp

@@ -6,6 +6,7 @@
 #include <mxl/flow.h>
 #include <mxl/time.h>
 #include "NodeBase.hpp"
+#include "FlowDef.hpp"
 #include <Processing.NDI.Lib.h>
 
 // RAII wrapper: init NDI, create sender, destroy both on scope exit.
@@ -45,14 +46,7 @@ class NDIOutNode : public dmf::NodeBase {
         uint32_t          video_stride = 0;
 
         if (has_video) {
-            const auto flow_info = config().at("video_flow_id");
+            flow_id = config().at("video_flow_id").at("id").get<std::string>();
-            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;
@@ -61,13 +55,21 @@ class NDIOutNode : public dmf::NodeBase {
                 if (!active) mxlSleepForNs(100'000'000);
             }
             if (!dmf::g_running) return;
-            log("flow active — starting read");
 
             mxlFlowConfigInfo video_cfg{};
             mxlStatus vst = mxlCreateFlowReader(instance(), flow_id.c_str(), "", &video_reader);
             if (vst != MXL_STATUS_OK) { log("video mxlCreateFlowReader failed (%s)", dmf::mxl_status_str(vst)); return; }
             mxlFlowReaderGetConfigInfo(video_reader, &video_cfg);
             video_stride = video_cfg.discrete.sliceSizes[0];
+
+            // Read actual format from what the source wrote — never trust config values here
+            const auto vfi = dmf::read_video_flow_info(domain(), flow_id);
+            width   = vfi.width;
+            height  = vfi.height;
+            fps_num = vfi.fps_num;
+            fps_den = vfi.fps_den;
+            log("video flow=%s  %dx%d @ %d/%d fps  stride=%u",
+                flow_id.c_str(), width, height, fps_num, fps_den, video_stride);
         }
 
         // --- audio flow (optional) ---
@@ -80,23 +82,32 @@ class NDIOutNode : public dmf::NodeBase {
         std::string audio_flow_id;
 
         if (has_audio) {
-            const auto audio_flow_info = config().at("audio_flow_id");
+            audio_flow_id = config().at("audio_flow_id").at("id").get<std::string>();
-            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;
 
-            log("audio flow=%s  %d Hz  %dch  %d samples/frame",
+            log("waiting for audio flow to become active...");
-                audio_flow_id.c_str(), sample_rate, channels, samples_per_frame);
+            bool active = false;
+            while (!active && dmf::g_running.load(std::memory_order_relaxed)) {
+                mxlIsFlowActive(instance(), audio_flow_id.c_str(), &active);
+                if (!active) mxlSleepForNs(100'000'000);
+            }
+            if (!dmf::g_running) { has_audio = false; }
+        }
 
+        if (has_audio) {
             mxlStatus ast = mxlCreateFlowReader(instance(), audio_flow_id.c_str(), "", &audio_reader);
             if (ast != MXL_STATUS_OK) {
                 log("audio mxlCreateFlowReader failed (%s) — continuing without audio", dmf::mxl_status_str(ast));
                 has_audio = false;
             } else {
                 mxlFlowReaderGetConfigInfo(audio_reader, &audio_cfg);
-                log("audio channels=%u  buffer=%u samples",
+                const auto afi = dmf::read_audio_flow_info(domain(), audio_flow_id);
-                    audio_cfg.continuous.channelCount, audio_cfg.continuous.bufferLength);
+                sample_rate       = afi.sample_rate;
+                channels          = afi.channels;
+                samples_per_frame = afi.samples_per_grain;
+                log("audio flow=%s  %d Hz  %dch  %d samples/frame",
+                    audio_flow_id.c_str(), sample_rate, channels, samples_per_frame);
+                log("audio mxl buffer=%u samples",
+                    audio_cfg.continuous.bufferLength);
             }
         }
 

nodes/videoin/main.cpp

@@ -30,10 +30,10 @@ class VideoInNode : public dmf::NodeBase {
         if (has_video) {
             const auto video_flow_info = config().at("video_flow_id");
             video_flow_id = video_flow_info.at("id").get<std::string>();
-            width    = video_flow_info.value("width",   video_reader.video_info.width);
+            width    = video_reader.video_info.width;
-            height   = video_flow_info.value("height",  video_reader.video_info.height);
+            height   = video_reader.video_info.height;
-            fps_num  = video_flow_info.value("fps_num", video_reader.video_info.fps_num);
+            fps_num  = video_reader.video_info.fps_num;
-            fps_den  = video_flow_info.value("fps_den", video_reader.video_info.fps_den);
+            fps_den  = video_reader.video_info.fps_den;
 
             log("video flow=%s  %dx%d @ %d/%d fps", video_flow_id.c_str(), width, height, fps_num, fps_den);
 

shared/FlowDef.hpp

@@ -1,4 +1,5 @@
 #pragma once
+#include <fstream>
 #include <string>
 #include <nlohmann/json.hpp>
 
@@ -72,4 +73,55 @@ inline std::string make_audio_flow_def(
     }.dump();
 }
 
+// Actual video format as written into the MXL flow_def.json by a source node.
+// Sink nodes use this to read width/height/fps from MXL instead of trusting config values.
+struct VideoFlowInfo {
+    int width   = 1920;
+    int height  = 1080;
+    int fps_num = 25;
+    int fps_den = 1;
+};
+
+inline VideoFlowInfo read_video_flow_info(const std::string& domain, const std::string& flow_id) {
+    VideoFlowInfo info;
+    const std::string path = domain + "/" + flow_id + ".mxl-flow/flow_def.json";
+    std::ifstream f(path);
+    if (!f.is_open()) return info;
+    const auto j = nlohmann::json::parse(f, nullptr, /*allow_exceptions=*/false);
+    if (j.is_discarded()) return info;
+    info.width   = j.value("frame_width",  1920);
+    info.height  = j.value("frame_height", 1080);
+    if (j.contains("grain_rate")) {
+        info.fps_num = j["grain_rate"].value("numerator",   25);
+        info.fps_den = j["grain_rate"].value("denominator",  1);
+    }
+    return info;
+}
+
+// Actual audio format as written into the MXL flow_def.json by a source node.
+struct AudioFlowInfo {
+    int sample_rate      = 48000;
+    int channels         = 2;
+    int samples_per_grain = 1920; // sample_rate * grain_rate_den / grain_rate_num
+};
+
+inline AudioFlowInfo read_audio_flow_info(const std::string& domain, const std::string& flow_id) {
+    AudioFlowInfo info;
+    const std::string path = domain + "/" + flow_id + ".mxl-flow/flow_def.json";
+    std::ifstream f(path);
+    if (!f.is_open()) return info;
+    const auto j = nlohmann::json::parse(f, nullptr, /*allow_exceptions=*/false);
+    if (j.is_discarded()) return info;
+    if (j.contains("sample_rate"))
+        info.sample_rate = j["sample_rate"].value("numerator", 48000);
+    info.channels = j.value("channel_count", 2);
+    if (j.contains("grain_rate") && info.sample_rate > 0) {
+        const int gr_num = j["grain_rate"].value("numerator",   25);
+        const int gr_den = j["grain_rate"].value("denominator",  1);
+        if (gr_num > 0)
+            info.samples_per_grain = info.sample_rate * gr_den / gr_num;
+    }
+    return info;
+}
+
 } // namespace dmf

shared/NDIReceiver.hpp

@@ -121,6 +121,39 @@ public:
         int channel_stride = 0;  // floats between channel planes (NDI planar layout)
     };
 
+    // Waits for the first audio packet and returns its format.
+    // Call after probe(). Audio packets are typically already queued at that point.
+    // Discards any video frames encountered while searching. Respects g_running.
+    AudioInfo probe_audio(uint32_t timeout_ms = 5000) {
+        const auto deadline = std::chrono::steady_clock::now()
+                              + std::chrono::milliseconds(timeout_ms);
+        while (dmf::g_running.load(std::memory_order_relaxed) &&
+               std::chrono::steady_clock::now() < deadline) {
+            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, 100);
+            if (type == NDIlib_frame_type_error)
+                throw std::runtime_error("NDI source lost during audio probe");
+            if (type == NDIlib_frame_type_video) {
+                NDIlib_recv_free_video_v2(recv_, &video_frame);
+                continue;
+            }
+            if (type == NDIlib_frame_type_audio) {
+                const bool is_fltp = (audio_frame.FourCC == NDIlib_FourCC_audio_type_FLTP);
+                AudioInfo info;
+                if (is_fltp) {
+                    info.sample_rate    = audio_frame.sample_rate;
+                    info.channels       = audio_frame.no_channels;
+                    info.samples        = audio_frame.no_samples;
+                    info.channel_stride = audio_frame.channel_stride_in_bytes / sizeof(float);
+                }
+                NDIlib_recv_free_audio_v3(recv_, &audio_frame);
+                if (is_fltp) return info;
+            }
+        }
+        throw std::runtime_error("Timeout waiting for first audio packet");
+    }
+
     // 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.