nodes/decklinkin/main.cpp

@@ -8,14 +8,17 @@
 
 class DeckLinkInNode : public dmf::NodeBase {
     void run() override {
-        const uint32_t device_index = config().value("device_index", 0u);
+        const uint32_t device_index  = config().value("device_index", 0u);
+        const bool     want_audio    = config().contains("audio_flow_id");
+        const int      channels      = want_audio
+            ? config().at("audio_flow_id").value("channels", 2) : 0;
 
         dmf::DeckLinkReceiver receiver;
         try {
             log("Available DeckLink devices:");
             for (const auto& d : receiver.devices)
                 log("  %u) %s", d.index, d.name.c_str());
-            receiver.start_capture(device_index);
+            receiver.start_capture(device_index, channels);
             log("Capturing from: %s", receiver.devices[device_index].name.c_str());
         } catch (const std::runtime_error& e) {
             log("DeckLink init error: %s", e.what());
@@ -30,13 +33,13 @@ class DeckLinkInNode : public dmf::NodeBase {
         if (vi.width == 0 || vi.fps_num == 0) { log("Invalid format detected"); return; }
         log("Detected: %dx%d @ %d/%d fps", vi.width, vi.height, vi.fps_num, vi.fps_den);
 
-        const auto video_flow_info = config().at("video_flow_id");
+        // --- Video writer ---
-        const auto video_flow_id   = video_flow_info.at("id").get<std::string>();
+        const auto  video_flow_info = config().at("video_flow_id");
-        const int  width   = video_flow_info.value("width",   vi.width);
+        const auto  video_flow_id   = video_flow_info.at("id").get<std::string>();
-        const int  height  = video_flow_info.value("height",  vi.height);
+        const int   width   = video_flow_info.value("width",   vi.width);
-        const int  fps_num = video_flow_info.value("fps_num", vi.fps_num);
+        const int   height  = video_flow_info.value("height",  vi.height);
-        const int  fps_den = video_flow_info.value("fps_den", vi.fps_den);
+        const int   fps_num = video_flow_info.value("fps_num", vi.fps_num);
-
+        const int   fps_den = video_flow_info.value("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;
@@ -50,36 +53,115 @@ class DeckLinkInNode : public dmf::NodeBase {
             log("mxlCreateFlowWriter failed (%s)", dmf::mxl_status_str(vst));
             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);
 
-        const mxlRational video_rate  = {fps_num, fps_den};
+        // --- Audio writer (optional) ---
-        uint64_t          video_index = mxlGetCurrentIndex(&video_rate);
+        mxlFlowWriter     audio_writer       = nullptr;
-        log("start video_index=%llu", video_index);
+        int               max_audio_samples  = 0;
+        const bool        has_audio          = want_audio && receiver.has_audio;
 
-        std::vector<uint8_t> frame_buf(static_cast<size_t>(video_stride) * static_cast<size_t>(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>();
+            const int  sample_rate     = receiver.audio_info.sample_rate;
+            const int  bit_depth       = 32;
+            log("audio flow=%s  %d Hz  %dch  %d-bit",
+                audio_flow_id.c_str(), sample_rate, channels, bit_depth);
 
-        while (dmf::g_running.load(std::memory_order_relaxed)) {
+            mxlFlowConfigInfo audio_cfg{};
-            if (!receiver.wait_for_frame(frame_buf.data(), video_stride, width, height)) break;
+            mxlStatus ast = mxlCreateFlowWriter(
-
+                instance(),
-            mxlGrainInfo grain{};
+                dmf::make_audio_flow_def(audio_flow_id, node_id(), sample_rate, channels, bit_depth,
-            uint8_t*     video_buf = nullptr;
+                                          fps_num, fps_den).c_str(),
-            vst = mxlFlowWriterOpenGrain(video_writer, video_index, &grain, &video_buf);
+                "", &audio_writer, &audio_cfg, &created);
-            if (vst == MXL_STATUS_OK) {
+            if (ast != MXL_STATUS_OK) {
-                std::memcpy(video_buf, frame_buf.data(), frame_buf.size());
+                log("audio mxlCreateFlowWriter failed (%s) — continuing without audio",
-                grain.flags       = 0;
+                    dmf::mxl_status_str(ast));
-                grain.validSlices = grain.totalSlices;
+            } else {
-                mxlFlowWriterCommitGrain(video_writer, &grain);
+                log("audio channels=%u  buffer=%u samples",
-                const uint64_t ns = mxlGetNsUntilIndex(video_index + 1, &video_rate);
+                    audio_cfg.continuous.channelCount, audio_cfg.continuous.bufferLength);
-                if (ns > 0 && ns < 2'000'000'000ULL) mxlSleepForNs(ns);
+                size_t max_write = 0;
-                video_index = mxlGetCurrentIndex(&video_rate);
+                mxlFlowWriterGetMaxWriteLengthSamples(audio_writer, &max_write);
+                max_audio_samples = static_cast<int>(max_write);
             }
         }
 
-        log("stopped at video_index=%llu", video_index);
+        // --- Buffers ---
+        std::vector<uint8_t> frame_buf(static_cast<size_t>(video_stride) * static_cast<size_t>(height));
+        // Planar float32: channel c at audio_buf[c * max_audio_samples]
+        std::vector<float>   audio_buf(static_cast<size_t>(max_audio_samples) * static_cast<size_t>(channels));
+
+        // --- Clock ---
+        const mxlRational video_rate  = {fps_num, fps_den};
+        const mxlRational audio_rate  = {receiver.audio_info.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", static_cast<unsigned long long>(video_index));
+
+        // --- Capture loop ---
+        while (dmf::g_running.load(std::memory_order_relaxed)) {
+            int samples_written = 0;
+
+            // DeckLink delivers one video frame + accompanying audio per callback.
+            if (!receiver.wait_for_frame(
+                    frame_buf.data(), video_stride, width, height,
+                    (has_audio && audio_writer) ? audio_buf.data() : nullptr,
+                    max_audio_samples,
+                    (has_audio && audio_writer) ? &samples_written : nullptr)) break;
+
+            // Video grain
+            mxlGrainInfo grain{};
+            uint8_t*     video_buf_ptr = nullptr;
+            vst = mxlFlowWriterOpenGrain(video_writer, video_index, &grain, &video_buf_ptr);
+            if (vst == MXL_STATUS_OK) {
+                std::memcpy(video_buf_ptr, frame_buf.data(), frame_buf.size());
+                grain.flags       = 0;
+                grain.validSlices = grain.totalSlices;
+                mxlFlowWriterCommitGrain(video_writer, &grain);
+            }
+
+            // Audio samples (same fragment-wrap pattern as videoin)
+            if (has_audio && audio_writer && samples_written > 0) {
+                mxlMutableWrappedMultiBufferSlice slice{};
+                mxlStatus ast = mxlFlowWriterOpenSamples(
+                    audio_writer, audio_index, static_cast<size_t>(samples_written), &slice);
+                if (ast == MXL_STATUS_OK) {
+                    for (int ch = 0; ch < channels; ++ch) {
+                        const uint8_t* src        = reinterpret_cast<const uint8_t*>(
+                            audio_buf.data() + ch * max_audio_samples);
+                        uint8_t*       dst0       = static_cast<uint8_t*>(
+                            slice.base.fragments[0].pointer) + ch * slice.stride;
+                        const size_t   frag0_bytes = slice.base.fragments[0].size;
+                        const size_t   total_bytes = static_cast<size_t>(samples_written) * sizeof(float);
+
+                        if (total_bytes <= frag0_bytes) {
+                            std::memcpy(dst0, src, total_bytes);
+                        } else {
+                            std::memcpy(dst0, src, frag0_bytes);
+                            uint8_t* dst1 = static_cast<uint8_t*>(
+                                slice.base.fragments[1].pointer) + ch * slice.stride;
+                            std::memcpy(dst1, src + frag0_bytes, total_bytes - frag0_bytes);
+                        }
+                    }
+                    mxlFlowWriterCommitSamples(audio_writer);
+                } else {
+                    log("audio OpenSamples failed (%s) index=%llu — skipping",
+                        dmf::mxl_status_str(ast), static_cast<unsigned long long>(audio_index));
+                }
+                audio_index += static_cast<uint64_t>(samples_written);
+            }
+
+            // Pace video to the MXL clock
+            const uint64_t ns = mxlGetNsUntilIndex(video_index + 1, &video_rate);
+            if (ns > 0 && ns < 2'000'000'000ULL) mxlSleepForNs(ns);
+            video_index = mxlGetCurrentIndex(&video_rate);
+        }
+
+        log("stopped at video_index=%llu", static_cast<unsigned long long>(video_index));
         mxlReleaseFlowWriter(instance(), video_writer);
+        if (audio_writer) mxlReleaseFlowWriter(instance(), audio_writer);
     }
 };
 

shared/DeckLinkReceiver.hpp

@@ -24,6 +24,11 @@ public:
         int fps_den = 0;
     };
 
+    struct AudioInfo {
+        int sample_rate = 48000;  // DeckLink always delivers 48 kHz
+        int channels    = 0;
+    };
+
     struct DeviceInfo {
         uint32_t    index;
         std::string name;
@@ -31,6 +36,8 @@ public:
 
     std::vector<DeviceInfo> devices;
     VideoInfo               video_info{};
+    AudioInfo               audio_info{};
+    bool                    has_audio = false;
 
     DeckLinkReceiver() { enumerate_devices(); }
 
@@ -38,6 +45,7 @@ public:
         if (decklink_input) {
             decklink_input->StopStreams();
             decklink_input->DisableVideoInput();
+            if (has_audio) decklink_input->DisableAudioInput();
             decklink_input->SetCallback(nullptr);
             decklink_input->Release();
         }
@@ -46,7 +54,8 @@ public:
         if (selected_device) selected_device->Release();
     }
 
-    void start_capture(uint32_t device_index) {
+    // audio_channels > 0 enables audio capture at 48 kHz / 32-bit int.
+    void start_capture(uint32_t device_index, int audio_channels = 0) {
         if (device_index >= raw_devices.size())
             throw std::runtime_error("Device index out of range");
 
@@ -64,6 +73,15 @@ public:
                                               bmdVideoInputEnableFormatDetection);
         if (r != S_OK) throw std::runtime_error("Could not enable video input");
 
+        if (audio_channels > 0) {
+            r = decklink_input->EnableAudioInput(bmdAudioSampleRate48kHz,
+                                                  bmdAudioSampleType32bitInteger,
+                                                  static_cast<uint32_t>(audio_channels));
+            if (r != S_OK) throw std::runtime_error("Could not enable audio input");
+            audio_info.channels = audio_channels;
+            has_audio = true;
+        }
+
         r = decklink_input->StartStreams();
         if (r != S_OK) throw std::runtime_error("Could not start streams");
     }
@@ -75,24 +93,40 @@ public:
     }
 
     // Blocks until a fresh frame arrives or g_running goes false.
-    // Copies frame data into dst with MXL stride. Returns false on shutdown.
+    // Copies video into video_dst and (if audio_dst != nullptr) deinterleaved
-    bool wait_for_frame(uint8_t* dst, uint32_t dst_stride, int width, int height) {
+    // float32 audio into audio_dst[channel * max_samples + sample].
+    // Returns false on shutdown.
+    bool wait_for_frame(uint8_t* video_dst, uint32_t dst_stride, int width, int height,
+                        float* audio_dst = nullptr, int max_samples = 0, int* samples_written = nullptr) {
         std::unique_lock<std::mutex> lk(mutex);
         frame_cv.wait(lk, [this] {
             return frame_ready || !dmf::g_running.load(std::memory_order_relaxed);
         });
         if (!dmf::g_running.load(std::memory_order_relaxed)) return false;
 
-        const uint32_t src_stride  = frame_row_bytes;
+        // Video copy
-        const int      rows        = std::min(frame_height, height);
+        const uint32_t src_stride = frame_row_bytes;
-        const uint32_t copy_bytes  = std::min(src_stride, dst_stride);
+        const int      rows       = std::min(frame_height, height);
-
+        const uint32_t copy_bytes = std::min(src_stride, dst_stride);
-        std::memset(dst, 0, static_cast<size_t>(dst_stride) * static_cast<size_t>(height));
+        std::memset(video_dst, 0, static_cast<size_t>(dst_stride) * static_cast<size_t>(height));
         const uint8_t* src = frame_buffer.data();
-        uint8_t*       d   = dst;
+        uint8_t*       d   = video_dst;
         for (int y = 0; y < rows; ++y, src += src_stride, d += dst_stride)
             std::memcpy(d, src, copy_bytes);
 
+        // Audio copy — planar float32: channel c starts at audio_dst + c * max_samples
+        if (audio_dst && max_samples > 0 && samples_written) {
+            const int n  = std::min(audio_samples, max_samples);
+            const int ch = audio_info.channels;
+            *samples_written = n;
+            for (int c = 0; c < ch; ++c)
+                std::memcpy(audio_dst + c * max_samples,
+                            audio_buffer.data() + c * audio_samples,
+                            static_cast<size_t>(n) * sizeof(float));
+        } else if (samples_written) {
+            *samples_written = 0;
+        }
+
         frame_ready = false;
         return true;
     }
@@ -121,7 +155,7 @@ private:
                     const int g = std::gcd(owner.video_info.fps_num, owner.video_info.fps_den);
                     if (g > 1) { owner.video_info.fps_num /= g; owner.video_info.fps_den /= g; }
                 }
-                owner.frame_ready    = false;
+                owner.frame_ready     = false;
                 owner.frame_buffer.clear();
                 owner.format_detected = true;
             }
@@ -139,7 +173,7 @@ private:
 
         HRESULT STDMETHODCALLTYPE VideoInputFrameArrived(
             IDeckLinkVideoInputFrame*  video_frame,
-            IDeckLinkAudioInputPacket* /*audio_packet*/) override
+            IDeckLinkAudioInputPacket* audio_packet) override
         {
             if (!video_frame) return S_OK;
 
@@ -157,12 +191,33 @@ private:
                 const size_t   sz        = static_cast<size_t>(row_bytes) * static_cast<size_t>(fh);
 
                 std::lock_guard<std::mutex> lk(owner.mutex);
+
+                // Video
                 if (owner.frame_buffer.size() < sz) owner.frame_buffer.resize(sz);
                 std::memcpy(owner.frame_buffer.data(), src, sz);
                 owner.frame_row_bytes = row_bytes;
                 owner.frame_width     = fw;
                 owner.frame_height    = fh;
-                owner.frame_ready     = true;
+
+                // Audio — deinterleave int32 → float32 planar under the same lock
+                if (owner.has_audio && audio_packet) {
+                    void* asrc = nullptr;
+                    audio_packet->GetBytes(&asrc);
+                    const long nb = audio_packet->GetSampleFrameCount();
+                    const int  ch = owner.audio_info.channels;
+                    if (asrc && nb > 0 && ch > 0) {
+                        const auto*  in   = static_cast<const int32_t*>(asrc);
+                        const size_t need = static_cast<size_t>(ch) * static_cast<size_t>(nb);
+                        if (owner.audio_buffer.size() < need) owner.audio_buffer.resize(need);
+                        for (long s = 0; s < nb; ++s)
+                            for (int c = 0; c < ch; ++c)
+                                owner.audio_buffer[static_cast<size_t>(c) * static_cast<size_t>(nb) + static_cast<size_t>(s)]
+                                    = static_cast<float>(in[s * ch + c]) / 2147483648.0f;
+                        owner.audio_samples = static_cast<int>(nb);
+                    }
+                }
+
+                owner.frame_ready = true;
             }
 
             buf->EndAccess(bmdBufferAccessRead);
@@ -199,6 +254,10 @@ private:
     int                     frame_height    = 0;
     bool                    frame_ready     = false;
 
+    // Audio state — planar float32: channel c at audio_buffer[c * audio_samples + s]
+    std::vector<float>      audio_buffer;
+    int                     audio_samples = 0;
+
     void enumerate_devices() {
         IDeckLinkIterator* it = CreateDeckLinkIteratorInstance();
         if (!it) throw std::runtime_error("DeckLink drivers not installed");