Merge pull request 'Decklink in' (#4) from decklink-in into main

Reviewed-on: #4

.vscode/c_cpp_properties.json

@@ -5,7 +5,8 @@
             "includePath": [
                 "${workspaceFolder}/**",
                 "${workspaceFolder}/shared",
-                "${HOME}/SDK/NDI/include"
+                "${HOME}/SDK/NDI/include",
+                "${HOME}/SDK/decklink-sdk/Linux/include"
             ],
             "defines": [],
             "compilerPath": "/usr/bin/clang",

CMakeLists.txt

@@ -106,6 +106,12 @@ if(NDI_SDK_DIR)
     add_subdirectory(nodes/ndiin)
 endif()
 
+# ── DeckLink nodes ────────────────────────────────────────────────────────────────
+set(DECKLINK_SDK_DIR "" CACHE PATH "Path to DeckLink SDK root")
+if(DECKLINK_SDK_DIR)
+    add_subdirectory(nodes/decklinkin)
+endif()
+
 add_subdirectory(nodes/videoin)
 
 # ── Core server ──────────────────────────────────────────────────────────────

decklink-ndi.json

@@ -0,0 +1,13 @@
+{
+  "nodes": [
+    { "id": "decklinkin", "type": "decklinkin", "params": {} },
+    { "id": "ndiout",      "type": "ndiout",      "params": {} }
+  ],
+  "edges": [
+    {
+      "from": "decklinkin",  "from_port": "video_flow_id",
+      "to":   "ndiout",      "to_port":   "video_flow_id",
+      "format": { "kind": "video", "width": 1920, "height": 1080, "fps_num": 25, "fps_den": 1 }
+    }
+  ]
+}

nodes/decklinkin/CMakeLists.txt

@@ -0,0 +1,11 @@
+set(DECKLINK_INCLUDE "${DECKLINK_SDK_DIR}/include"
+    CACHE PATH "Path to DeckLink API headers")
+
+add_executable(dmf-node-decklinkin
+    main.cpp
+    "${DECKLINK_INCLUDE}/DeckLinkAPIDispatch.cpp"
+)
+target_compile_features(dmf-node-decklinkin PRIVATE cxx_std_20)
+target_include_directories(dmf-node-decklinkin PRIVATE "${DECKLINK_INCLUDE}")
+target_link_libraries(dmf-node-decklinkin PRIVATE dmf-shared ${CMAKE_DL_LIBS})
+install(TARGETS dmf-node-decklinkin RUNTIME DESTINATION bin)

nodes/decklinkin/main.cpp

@@ -0,0 +1,171 @@
+#include <cstring>
+#include <vector>
+#include <mxl/flow.h>
+#include <mxl/time.h>
+#include "NodeBase.hpp"
+#include "FlowDef.hpp"
+#include "DeckLinkReceiver.hpp"
+
+class DeckLinkInNode : public dmf::NodeBase {
+    void run() override {
+        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, 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());
+            return;
+        }
+
+        if (!receiver.wait_for_format(5000)) {
+            log("Timeout waiting for format detection");
+            return;
+        }
+        const auto& vi = receiver.video_info;
+        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);
+
+        // --- Video writer ---
+        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   height  = video_flow_info.value("height",  vi.height);
+        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;
+        mxlFlowConfigInfo video_cfg{};
+        bool created = false;
+        mxlStatus vst = mxlCreateFlowWriter(
+            instance(),
+            dmf::make_video_flow_def(video_flow_id, node_id(), width, height, fps_num, fps_den).c_str(),
+            "", &video_writer, &video_cfg, &created);
+        if (vst != MXL_STATUS_OK) {
+            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);
+
+        // --- Audio writer (optional) ---
+        mxlFlowWriter     audio_writer       = nullptr;
+        int               max_audio_samples  = 0;
+        const bool        has_audio          = want_audio && receiver.has_audio;
+
+        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);
+
+            mxlFlowConfigInfo audio_cfg{};
+            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(),
+                "", &audio_writer, &audio_cfg, &created);
+            if (ast != MXL_STATUS_OK) {
+                log("audio mxlCreateFlowWriter failed (%s) — continuing without audio",
+                    dmf::mxl_status_str(ast));
+            } else {
+                log("audio channels=%u  buffer=%u samples",
+                    audio_cfg.continuous.channelCount, audio_cfg.continuous.bufferLength);
+                size_t max_write = 0;
+                mxlFlowWriterGetMaxWriteLengthSamples(audio_writer, &max_write);
+                max_audio_samples = static_cast<int>(max_write);
+            }
+        }
+
+        // --- 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);
+    }
+};
+
+int main() {
+    DeckLinkInNode node;
+    return node.execute();
+}

shared/DeckLinkReceiver.hpp

@@ -0,0 +1,286 @@
+#pragma once
+
+#include <algorithm>
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <cstring>
+#include <mutex>
+#include <numeric>
+#include <stdexcept>
+#include <string>
+#include <vector>
+#include <DeckLinkAPI.h>
+#include "Signal.hpp"
+
+namespace dmf {
+
+class DeckLinkReceiver {
+public:
+    struct VideoInfo {
+        int width   = 0;
+        int height  = 0;
+        int fps_num = 0;
+        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;
+    };
+
+    std::vector<DeviceInfo> devices;
+    VideoInfo               video_info{};
+    AudioInfo               audio_info{};
+    bool                    has_audio = false;
+
+    DeckLinkReceiver() { enumerate_devices(); }
+
+    ~DeckLinkReceiver() {
+        if (decklink_input) {
+            decklink_input->StopStreams();
+            decklink_input->DisableVideoInput();
+            if (has_audio) decklink_input->DisableAudioInput();
+            decklink_input->SetCallback(nullptr);
+            decklink_input->Release();
+        }
+        delete input_callback;
+        for (auto* d : raw_devices) d->Release();
+        if (selected_device) selected_device->Release();
+    }
+
+    // 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");
+
+        selected_device = raw_devices[device_index];
+        selected_device->AddRef();
+
+        HRESULT r = selected_device->QueryInterface(IID_IDeckLinkInput, (void**)&decklink_input);
+        if (r != S_OK) throw std::runtime_error("Could not obtain IDeckLinkInput");
+
+        input_callback = new InputCallback(*this);
+        r = decklink_input->SetCallback(input_callback);
+        if (r != S_OK) throw std::runtime_error("Could not set input callback");
+
+        r = decklink_input->EnableVideoInput(bmdModeNTSC, bmdFormat10BitYUV,
+                                              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");
+    }
+
+    bool wait_for_format(int timeout_ms = 5000) {
+        std::unique_lock<std::mutex> lk(mutex);
+        return format_cv.wait_for(lk, std::chrono::milliseconds(timeout_ms),
+            [this] { return format_detected; });
+    }
+
+    // Blocks until a fresh frame arrives or g_running goes false.
+    // Copies video into video_dst and (if audio_dst != nullptr) deinterleaved
+    // 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;
+
+        // Video copy
+        const uint32_t src_stride = frame_row_bytes;
+        const int      rows       = std::min(frame_height, height);
+        const uint32_t copy_bytes = std::min(src_stride, dst_stride);
+        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   = 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;
+    }
+
+private:
+    class InputCallback : public IDeckLinkInputCallback {
+    public:
+        explicit InputCallback(DeckLinkReceiver& owner) : owner(owner) {}
+
+        HRESULT STDMETHODCALLTYPE VideoInputFormatChanged(
+            BMDVideoInputFormatChangedEvents events,
+            IDeckLinkDisplayMode*            mode,
+            BMDDetectedVideoInputFormatFlags /*flags*/) override
+        {
+            if (!(events & bmdVideoInputDisplayModeChanged)) return S_OK;
+
+            {
+                std::lock_guard<std::mutex> lk(owner.mutex);
+                owner.video_info.width  = static_cast<int>(mode->GetWidth());
+                owner.video_info.height = static_cast<int>(mode->GetHeight());
+                BMDTimeValue dur = 0; BMDTimeScale ts = 0;
+                mode->GetFrameRate(&dur, &ts);
+                if (dur > 0 && ts > 0) {
+                    owner.video_info.fps_num = static_cast<int>(ts);
+                    owner.video_info.fps_den = static_cast<int>(dur);
+                    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_buffer.clear();
+                owner.format_detected = true;
+            }
+            owner.format_cv.notify_one();
+
+            HRESULT r = owner.decklink_input->StopStreams();
+            if (r == S_OK) {
+                r = owner.decklink_input->EnableVideoInput(
+                    mode->GetDisplayMode(), bmdFormat10BitYUV,
+                    bmdVideoInputEnableFormatDetection);
+                if (r == S_OK) owner.decklink_input->StartStreams();
+            }
+            return S_OK;
+        }
+
+        HRESULT STDMETHODCALLTYPE VideoInputFrameArrived(
+            IDeckLinkVideoInputFrame*  video_frame,
+            IDeckLinkAudioInputPacket* audio_packet) override
+        {
+            if (!video_frame) return S_OK;
+
+            IDeckLinkVideoBuffer* buf = nullptr;
+            if (video_frame->QueryInterface(IID_IDeckLinkVideoBuffer, (void**)&buf) != S_OK)
+                return S_OK;
+            buf->StartAccess(bmdBufferAccessRead);
+            void* src = nullptr;
+            buf->GetBytes(&src);
+
+            if (src) {
+                const uint32_t row_bytes = static_cast<uint32_t>(video_frame->GetRowBytes());
+                const int      fw        = static_cast<int>(video_frame->GetWidth());
+                const int      fh        = static_cast<int>(video_frame->GetHeight());
+                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;
+
+                // 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);
+            buf->Release();
+            if (src) owner.frame_cv.notify_one();
+            return S_OK;
+        }
+
+        HRESULT STDMETHODCALLTYPE QueryInterface(REFIID, LPVOID*) override { return E_NOINTERFACE; }
+        ULONG   STDMETHODCALLTYPE AddRef()  override { return ++ref_count; }
+        ULONG   STDMETHODCALLTYPE Release() override { return --ref_count; }
+
+    private:
+        DeckLinkReceiver&    owner;
+        std::atomic<int32_t> ref_count{1};
+    };
+
+    // DeckLink SDK objects
+    std::vector<IDeckLink*>  raw_devices;
+    IDeckLink*               selected_device = nullptr;
+    IDeckLinkInput*          decklink_input  = nullptr;
+    InputCallback*           input_callback  = nullptr;
+
+    // Synchronisation — one mutex guards all shared state below
+    std::mutex              mutex;
+    std::condition_variable format_cv;   // signalled when format is detected
+    std::condition_variable frame_cv;    // signalled when a frame arrives
+    bool                    format_detected = false;
+
+    // Frame state — written by InputCallback, read by wait_for_frame (both under mutex)
+    std::vector<uint8_t>    frame_buffer;
+    uint32_t                frame_row_bytes = 0;
+    int                     frame_width     = 0;
+    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");
+
+        IDeckLink* device = nullptr;
+        uint32_t   index  = 0;
+        while (it->Next(&device) == S_OK) {
+            IDeckLinkInput* inp = nullptr;
+            if (device->QueryInterface(IID_IDeckLinkInput, (void**)&inp) == S_OK) {
+                inp->Release();
+                const char* name = nullptr;
+                device->GetDisplayName(&name);
+                devices.push_back({index, name ? name : "?"});
+                raw_devices.push_back(device);
+                index++;
+            } else {
+                device->Release();
+            }
+        }
+        it->Release();
+        if (raw_devices.empty())
+            throw std::runtime_error("No DeckLink input devices found");
+    }
+};
+
+} // namespace dmf