Merge pull request 'Feature/decklink out' (#5) from feature/decklink-out into main

Reviewed-on: #5

CMakeLists.txt

@@ -102,14 +102,15 @@ add_subdirectory(nodes/fakesink)
 # ── NDI nodes ────────────────────────────────────────────────────────────────
 set(NDI_SDK_DIR "" CACHE PATH "Path to NDI SDK root")
 if(NDI_SDK_DIR)
-    add_subdirectory(nodes/ndiout)
     add_subdirectory(nodes/ndiin)
+    add_subdirectory(nodes/ndiout)
 endif()
 
 # ── DeckLink nodes ────────────────────────────────────────────────────────────────
 set(DECKLINK_SDK_DIR "" CACHE PATH "Path to DeckLink SDK root")
 if(DECKLINK_SDK_DIR)
     add_subdirectory(nodes/decklinkin)
+    add_subdirectory(nodes/decklinkout)
 endif()
 
 add_subdirectory(nodes/videoin)

PLAN.md

@@ -0,0 +1,122 @@
+# DMF Studio — Roadmap
+
+## Next steps (in order)
+
+### 1. WebSocket API in studio-manager
+Allow the graph to be changed at runtime without restarting.
+- Add a WebSocket server to `studio-manager`
+- API: load/reload graph, start/stop individual nodes, query status
+- `studio-manager` already has `load_graph()` — the WS layer calls it on demand
+  and diffs against the running set (stop removed nodes, fork new ones)
+- **Required before**: frontend, live source switching, PiP (otherwise every
+  graph change is a full restart)
+
+### 2. Processing nodes — PiP / mixer
+First node that takes multiple input flows and produces an output flow.
+- Uses `mxlFlowSynchronizationGroup` to align grains from two inputs
+- Reference implementation: `nodes/testpattern` (writer) + `nodes/fakesink` (reader)
+- Only becomes useful with the WebSocket API (so you can switch sources live)
+
+### 3. Vue.js frontend
+Visual graph editor that drives the WebSocket API.
+
+---
+
+## Redundancy
+
+Key constraint: **one writer per MXL flow** — can't run two identical nodes writing
+the same flow simultaneously. Redundancy lives at the pipeline level, not the node level.
+
+### Dual pipeline on separate machines
+
+```
+Machine 1 (k8s node A)                    Machine 2 (k8s node B)
+  decklinkin → [MXL] → ndiout              decklinkin → [MXL] → ndiout
+       ↓                                          ↓
+  (primary path)                            (backup path)
+         \                                       /
+          └──────→  [selector node]  ←──────────┘
+                          ↓
+                       ndiout (final)
+```
+
+**Selector node** — reads two input flows, monitors grain validity flags, switches
+to backup when primary fails. Fits the existing node model; uses
+`mxlFlowSynchronizationGroup` to watch both flows. Key processing node to build
+once redundancy becomes a requirement.
+
+MXL shared memory requires all pods in a pipeline to be co-located on the same
+physical machine. Redundant pipelines naturally go on *different* machines — which
+is exactly right for hardware failure redundancy.
+
+### What k8s gives for free
+
+- Stateless processing nodes (PiP, denoise, format convert): k8s restarts on crash,
+  ~1-2 s gap — acceptable for non-critical path
+- `PodDisruptionBudget`: ensures critical nodes survive cluster maintenance
+- Leader election (k8s lease objects): two studio-managers, one active, one standby;
+  automatic failover with no node code changes
+
+---
+
+## Kubernetes integration (mxl-k8s)
+
+Source: `~/codeproj/mxl-k8s` — not official, treat as reference, not truth.
+
+mxl-k8s is a full k8s control plane for MXL flows. Four runtime pieces:
+
+- **Operator** (Deployment): watches `MxlReceiver` CRDs, creates `MxlFlowMirror` per target node
+- **Agent** (DaemonSet): watches each node's MXL domain via `fanotify`, publishes `MxlFlow` CRDs with where flows live
+- **Gateway** (DaemonSet, `hostNetwork`): drives libmxl-fabrics RDMA/TCP between nodes — zero-copy grain transfer via registered mmap regions
+- **Shim** (`libmxl-intent.so`, LD_PRELOAD): intercepts `openat`/`stat`/`access` on `.mxl-flow/` paths in consumer pods; when a flow isn't local, asks the agent's UDS socket (`/run/mxl/agent.sock`) to materialize it via mirror, then retries — transparent to node code
+
+### What changes for our nodes in k8s
+
+**Producer pods** (decklinkin, ndiin, testpattern): **zero code change**.
+- Add `hostPath: /run/mxl/domain` volume + `IPC_LOCK`, `SYS_RESOURCE` capabilities
+- `NODE_CONFIG` → Pod env var (from ConfigMap)
+- `MXL_DOMAIN` → `/run/mxl/domain` (standardized in k8s context)
+
+**Consumer pods, same node**: same as above, no code change.
+
+**Consumer pods, different node**: still no code change.
+- Add `initContainer` copying `libmxl-intent.so` from shim image
+- Set `LD_PRELOAD=/opt/mxl-intent/libmxl-intent.so`
+- Mount `/run/mxl` (whole dir, not just `/domain`) so agent socket is accessible
+- Create an `MxlReceiver` CRD pointing at the flow — operator handles the mirror plumbing
+
+### What studio-manager becomes in k8s
+
+Currently: fork/exec child processes. In k8s: apply/delete Pods (or Deployments) with `NODE_CONFIG` env vars. For cross-node flows: create `MxlReceiver` CRDs instead of wiring flows manually.
+
+Same-node pipeline: all pods get `nodeAffinity: requiredDuringScheduling → same host`.
+Cross-node: add shim + `MxlReceiver`; mxl-k8s handles the rest.
+
+---
+
+## Architecture decisions
+
+### Separate audio and video threads in nodes
+
+**Decision**: sink nodes (`decklinkout`, `ndiout`) and likely source nodes should
+process audio and video on separate threads.
+
+**Why**: audio and video have different timing granularities.
+- Video: one grain every ~40 ms (at 25 fps) — coarse, can block
+- Audio: must flow continuously at sample granularity — any stall causes dropout
+
+In the current single-thread model, video stalls (e.g. `TOO_EARLY` retries)
+pause audio too. In `decklinkout` this is especially bad — DeckLink's timestamped
+audio buffer underruns if it isn't fed consistently.
+
+**What it looks like**:
+- **Audio thread**: tight loop, continuously drains MXL audio ring buffer and
+  pushes to output (DeckLink `ScheduleAudioSamples` / NDI send). No video logic.
+- **Video thread**: current main loop, handles grain read → process → output
+  at frame rate.
+- **Shared state**: only `g_running` and the output handle. No frame data crosses
+  the boundary — each thread reads its own MXL flow independently.
+  MXL clock keeps them in sync without explicit A/V coordination.
+
+**When**: after the WebSocket API, when running real content and audio quality matters.
+Current single-thread model is acceptable for development.

ndi-decklink.json

@@ -0,0 +1,18 @@
+{
+  "nodes": [
+    { "id": "ndiin",       "type": "ndiin",            "params": {} },
+    { "id": "decklinkout", "type": "decklinkout",      "params": { "device_index": 0 } }
+  ],
+  "edges": [
+    {
+      "from": "ndiin",  "from_port": "video_flow_id",
+      "to":   "decklinkout",      "to_port":   "video_flow_id",
+      "format": { "kind": "video", "width": 1920, "height": 1080, "fps_num": 25, "fps_den": 1 }
+    },
+    {
+      "from": "ndiin", "from_port": "audio_flow_id",
+      "to":   "decklinkout", "to_port":   "audio_flow_id",
+      "format": { "kind": "audio", "sample_rate": 48000, "channels": 2, "bit_depth": 32 }
+    }
+  ]
+}

ndi-ndi.json

@@ -0,0 +1,13 @@
+{
+  "nodes": [
+    { "id": "ndiin",       "type": "ndiin",            "params": {} },
+    { "id": "ndiout", "type": "ndiout",      "params": {} }
+  ],
+  "edges": [
+    {
+      "from": "ndiin",  "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/decklinkout/CMakeLists.txt

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

nodes/decklinkout/main.cpp

@@ -0,0 +1,230 @@
+#include <chrono>
+#include <cstring>
+#include <string>
+#include <vector>
+#include <mxl/flow.h>
+#include <mxl/time.h>
+#include "NodeBase.hpp"
+#include "DeckLinkSender.hpp"
+
+class DeckLinkOutNode : public dmf::NodeBase {
+    void run() override {
+        const uint32_t device_index = config().value("device_index", 0u);
+
+        // --- video flow (optional) ---
+        bool has_video = config().contains("video_flow_id");
+
+        int         width   = 1920;
+        int         height  = 1080;
+        int         fps_num = 25;
+        int         fps_den = 1;
+        std::string flow_id;
+        mxlFlowReader video_reader{};
+        uint32_t      video_stride = 0;
+
+        if (has_video) {
+            const auto flow_info = config().at("video_flow_id");
+            flow_id = flow_info.at("id").get<std::string>();
+            width   = flow_info.value("width",   1920);
+            height  = flow_info.value("height",  1080);
+            fps_num = flow_info.value("fps_num", 25);
+            fps_den = flow_info.value("fps_den", 1);
+            log("video flow=%s  %dx%d @ %d/%d fps", flow_id.c_str(), width, height, fps_num, fps_den);
+
+            log("waiting for flow to become active...");
+            bool active = false;
+            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");
+
+            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];
+        }
+
+        // --- audio flow (optional) ---
+        mxlFlowReader audio_reader{};
+        int  sample_rate       = 48000;
+        int  channels          = 0;
+        int  samples_per_frame = 0;
+        bool has_audio         = config().contains("audio_flow_id");
+        std::string audio_flow_id;
+
+        if (has_audio) {
+            const auto audio_flow_info = config().at("audio_flow_id");
+            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);
+
+            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 {
+                mxlFlowConfigInfo audio_cfg{};
+                mxlFlowReaderGetConfigInfo(audio_reader, &audio_cfg);
+                log("audio channels=%u  buffer=%u samples",
+                    audio_cfg.continuous.channelCount, audio_cfg.continuous.bufferLength);
+            }
+        }
+
+        if (!has_video && !has_audio) { log("no flows configured — exiting"); return; }
+
+        dmf::DeckLinkSender sender;
+        try {
+            log("Available DeckLink output devices:");
+            for (const auto& d : sender.devices)
+                log("  %u) %s", d.index, d.name.c_str());
+            sender.start_output(device_index, width, height, fps_num, fps_den, channels);
+        } catch (const std::runtime_error& e) {
+            log("DeckLink init error: %s", e.what());
+            if (has_video  && video_reader) mxlReleaseFlowReader(instance(), video_reader);
+            if (has_audio  && audio_reader) mxlReleaseFlowReader(instance(), audio_reader);
+            return;
+        }
+
+        // Pre-allocate audio staging buffer (planar float32)
+        std::vector<float> audio_planar(
+            static_cast<size_t>(channels) * static_cast<size_t>(samples_per_frame));
+
+        // --- Clock init ---
+        uint64_t video_index = 0;
+        uint64_t audio_index = 0;
+        if (has_video) {
+            const mxlRational video_rate = {fps_num, fps_den};
+            video_index = mxlGetCurrentIndex(&video_rate);
+        }
+        if (has_audio) {
+            const mxlRational audio_rate = {sample_rate, 1};
+            audio_index = mxlGetCurrentIndex(&audio_rate);
+        }
+
+        uint64_t frame_count   = 0;
+        uint64_t invalid_count = 0;
+        uint64_t late_count    = 0;
+        auto wall_start    = std::chrono::steady_clock::now();
+        auto last_log_time = wall_start;
+
+        // --- Main loop ---
+        while (dmf::g_running.load(std::memory_order_relaxed)) {
+            // Audio: non-blocking, one chunk per video frame
+            bool audio_advanced = false;
+            if (has_audio) {
+                mxlWrappedMultiBufferSlice audio_slices{};
+                mxlStatus ast = mxlFlowReaderGetSamplesNonBlocking(
+                    audio_reader, audio_index, samples_per_frame, &audio_slices);
+                if (ast == MXL_STATUS_OK) {
+                    const size_t frag0 = audio_slices.base.fragments[0].size / sizeof(float);
+                    const size_t frag1 = audio_slices.base.fragments[1].size / sizeof(float);
+                    for (int c = 0; c < channels; ++c) {
+                        float* dst = audio_planar.data() + c * samples_per_frame;
+                        const auto* src0 = reinterpret_cast<const float*>(
+                            static_cast<const uint8_t*>(audio_slices.base.fragments[0].pointer)
+                            + static_cast<size_t>(c) * audio_slices.stride);
+                        std::memcpy(dst, src0, frag0 * sizeof(float));
+                        if (frag1 > 0) {
+                            const auto* src1 = reinterpret_cast<const float*>(
+                                static_cast<const uint8_t*>(audio_slices.base.fragments[1].pointer)
+                                + static_cast<size_t>(c) * audio_slices.stride);
+                            std::memcpy(dst + frag0, src1, frag1 * sizeof(float));
+                        }
+                    }
+                    sender.submit_audio(audio_planar.data(), samples_per_frame);
+                    audio_index  += samples_per_frame;
+                    audio_advanced = true;
+                } else if (ast == MXL_ERR_OUT_OF_RANGE_TOO_LATE) {
+                    mxlFlowRuntimeInfo ari{};
+                    mxlFlowReaderGetRuntimeInfo(audio_reader, &ari);
+                    audio_index = ari.headIndex;
+                } else if (ast == MXL_ERR_FLOW_INVALID) {
+                    log("audio flow invalidated — reconnecting...");
+                    mxlReleaseFlowReader(instance(), audio_reader);
+                    audio_reader = nullptr;
+                    mxlSleepForNs(100'000'000);
+                    if (mxlCreateFlowReader(instance(), audio_flow_id.c_str(), "", &audio_reader) == MXL_STATUS_OK) {
+                        log("audio flow reconnected");
+                        const mxlRational r = {sample_rate, 1};
+                        audio_index = mxlGetCurrentIndex(&r);
+                    }
+                }
+            }
+
+            // Video
+            if (has_video) {
+                mxlGrainInfo video_grain{};
+                uint8_t*     video_buf = nullptr;
+                mxlStatus vst = mxlFlowReaderGetGrainNonBlocking(
+                    video_reader, video_index, &video_grain, &video_buf);
+
+                if (vst == MXL_STATUS_OK) {
+                    if (video_grain.flags & MXL_GRAIN_FLAG_INVALID) invalid_count++;
+                    sender.submit_frame(video_buf, video_stride);
+                    frame_count++;
+                    video_index++;
+                } else if (vst == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) {
+                    mxlSleepForNs(1'000'000);
+                } else if (vst == MXL_ERR_OUT_OF_RANGE_TOO_LATE) {
+                    late_count++;
+                    mxlFlowRuntimeInfo ri{};
+                    mxlFlowReaderGetRuntimeInfo(video_reader, &ri);
+                    video_index = ri.headIndex;
+                } else if (vst == MXL_ERR_FLOW_INVALID) {
+                    log("video flow invalidated — reconnecting...");
+                    mxlReleaseFlowReader(instance(), video_reader);
+                    video_reader = nullptr;
+                    mxlSleepForNs(100'000'000);
+                    if (mxlCreateFlowReader(instance(), flow_id.c_str(), "", &video_reader) == MXL_STATUS_OK) {
+                        log("video flow reconnected");
+                        const mxlRational r = {fps_num, fps_den};
+                        video_index = mxlGetCurrentIndex(&r);
+                    }
+                } else {
+                    log("unexpected video status (%s) on index=%llu",
+                        dmf::mxl_status_str(vst), static_cast<unsigned long long>(video_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",
+                        static_cast<unsigned long long>(frame_count),
+                        static_cast<unsigned long long>(invalid_count),
+                        static_cast<unsigned long long>(late_count),
+                        static_cast<double>(frame_count) / elapsed);
+                    last_log_time = now;
+                }
+            } else if (!audio_advanced) {
+                mxlSleepForNs(1'000'000);
+            }
+        }
+
+        if (has_video)
+            log("stopped — total frames=%llu  invalid=%llu  late=%llu",
+                static_cast<unsigned long long>(frame_count),
+                static_cast<unsigned long long>(invalid_count),
+                static_cast<unsigned long long>(late_count));
+        else
+            log("stopped");
+
+        if (has_video && video_reader) mxlReleaseFlowReader(instance(), video_reader);
+        if (has_audio && audio_reader) mxlReleaseFlowReader(instance(), audio_reader);
+    }
+};
+
+int main() {
+    DeckLinkOutNode node;
+    return node.execute();
+}

nodes/ndiin/main.cpp

@@ -98,7 +98,7 @@ class NDIInNode : public dmf::NodeBase {
         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);
+                kind = ndi.capture(latest_video.data(), video_stride, audio_buf, audio_info, has_audio);
             } catch (const std::runtime_error& e) {
                 log("NDI error: %s — stopping", e.what());
                 break;
@@ -147,13 +147,13 @@ class NDIInNode : public dmf::NodeBase {
                     mxlFlowWriterCommitGrain(video_writer, &grain);
                     const uint64_t ns = mxlGetNsUntilIndex(video_index + 1, &video_rate);
                     if (ns > 0 && ns < 2'000'000'000ULL) mxlSleepForNs(ns);
-                    video_index++;
+                    video_index = mxlGetCurrentIndex(&video_rate);
                 } else {
                     log("video OpenGrain failed (%s) at index=%llu", dmf::mxl_status_str(st), video_index);
-                }
                     video_index = current + 1;
                 }
             }
+        }
 
         log("stopped at video_index=%llu", video_index);
         mxlReleaseFlowWriter(instance(), video_writer);

shared/DeckLinkSender.hpp

@@ -0,0 +1,288 @@
+#pragma once
+
+#include <algorithm>
+#include <atomic>
+#include <cmath>
+#include <condition_variable>
+#include <cstring>
+#include <mutex>
+#include <stdexcept>
+#include <string>
+#include <vector>
+#include <DeckLinkAPI.h>
+
+#include "Signal.hpp"
+#include "V210.hpp"
+
+namespace dmf {
+
+class DeckLinkSender {
+public:
+    struct VideoInfo {
+        int width   = 0;
+        int height  = 0;
+        int fps_num = 0;
+        int fps_den = 0;
+    };
+
+    struct AudioInfo {
+        int sample_rate = 48000;
+        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;
+
+    DeckLinkSender() { enumerate_devices(); }
+
+    ~DeckLinkSender() {
+        if (decklink_output) {
+            decklink_output->StopScheduledPlayback(0, nullptr, 1);
+            decklink_output->DisableVideoOutput();
+            if (has_audio) decklink_output->DisableAudioOutput();
+            decklink_output->SetScheduledFrameCompletionCallback(nullptr);
+            decklink_output->Release();
+        }
+        delete output_callback;
+        for (auto* vf : all_frames) vf->Release();
+        for (auto* d : raw_devices) d->Release();
+        if (selected_device) selected_device->Release();
+    }
+
+    // channels > 0 enables audio output at 48 kHz / 32-bit int.
+    void start_output(uint32_t device_index, int width, int height,
+                      int fps_num, int fps_den, int channels = 0) {
+        if (device_index >= raw_devices.size())
+            throw std::runtime_error("Device index out of range");
+
+        video_info = {width, height, fps_num, fps_den};
+
+        selected_device = raw_devices[device_index];
+        selected_device->AddRef();
+
+        HRESULT r = selected_device->QueryInterface(IID_IDeckLinkOutput, (void**)&decklink_output);
+        if (r != S_OK) throw std::runtime_error("Could not obtain IDeckLinkOutput");
+
+        output_callback = new OutputCallback(*this);
+        r = decklink_output->SetScheduledFrameCompletionCallback(output_callback);
+        if (r != S_OK) throw std::runtime_error("Could not set output callback");
+
+        const BMDDisplayMode mode = pick_display_mode(width, height, fps_num, fps_den);
+
+        // Verify the card supports this mode in 10-bit YUV before enabling
+        bool is_supported = false;
+        BMDDisplayMode actual_mode = mode;
+        r = decklink_output->DoesSupportVideoMode(
+            bmdVideoConnectionUnspecified,
+            mode,
+            bmdFormat10BitYUV,
+            bmdNoVideoOutputConversion,
+            bmdSupportedVideoModeDefault,
+            &actual_mode,
+            &is_supported);
+        if (r != S_OK || !is_supported)
+            throw std::runtime_error("Display mode not supported in 10-bit YUV");
+
+        r = decklink_output->EnableVideoOutput(actual_mode, bmdVideoOutputFlagDefault);
+        if (r != S_OK) throw std::runtime_error("Could not enable video output");
+
+        if (channels > 0) {
+            r = decklink_output->EnableAudioOutput(bmdAudioSampleRate48kHz,
+                                                    bmdAudioSampleType32bitInteger,
+                                                    static_cast<uint32_t>(channels),
+                                                    bmdAudioOutputStreamTimestamped);
+            if (r != S_OK) throw std::runtime_error("Could not enable audio output");
+            audio_info.channels = channels;
+            has_audio = true;
+        }
+
+        int32_t row_bytes = 0;
+        r = decklink_output->RowBytesForPixelFormat(bmdFormat10BitYUV, width, &row_bytes);
+        if (r != S_OK) throw std::runtime_error("Could not get row bytes for pixel format");
+
+        // Pre-fill 3 black frames and schedule them to prime the pipeline.
+        // The card fires ScheduledFrameCompleted once each is done, at which point
+        // submit_frame() can reclaim and reschedule them with live video.
+        constexpr size_t kPreroll = 3;
+        for (size_t i = 0; i < kPreroll; ++i) {
+            IDeckLinkMutableVideoFrame* vf = nullptr;
+            r = decklink_output->CreateVideoFrame(
+                width, height, row_bytes, bmdFormat10BitYUV, bmdFrameFlagDefault, &vf);
+            if (r != S_OK) throw std::runtime_error("Could not create video frame");
+
+            IDeckLinkVideoBuffer* buf = nullptr;
+            vf->QueryInterface(IID_IDeckLinkVideoBuffer, (void**)&buf);
+            buf->StartAccess(bmdBufferAccessWrite);
+            void* ptr = nullptr;
+            buf->GetBytes(&ptr);
+            if (ptr) dmf::v210::fill_black(static_cast<uint8_t*>(ptr), width, height, row_bytes);
+            buf->EndAccess(bmdBufferAccessWrite);
+            buf->Release();
+
+            decklink_output->ScheduleVideoFrame(
+                vf, static_cast<BMDTimeValue>(i * fps_den), fps_den, fps_num);
+            all_frames.push_back(vf);
+        }
+        scheduled_time = static_cast<int64_t>(kPreroll) * fps_den;
+
+        r = decklink_output->StartScheduledPlayback(0, fps_num, 1.0);
+        if (r != S_OK) throw std::runtime_error("Could not start scheduled playback");
+    }
+
+    // Blocks until a frame slot is free (returned by the card via callback),
+    // then copies src into it and schedules it for display.
+    void submit_frame(const uint8_t* src, uint32_t stride) {
+        IDeckLinkMutableVideoFrame* vf = nullptr;
+        {
+            std::unique_lock<std::mutex> lk(pool_mutex);
+            pool_cv.wait(lk, [this] {
+                return !free_frames.empty() ||
+                       !dmf::g_running.load(std::memory_order_relaxed);
+            });
+            if (free_frames.empty()) return;
+            vf = free_frames.back();
+            free_frames.pop_back();
+        }
+
+        IDeckLinkVideoBuffer* buf = nullptr;
+        if (vf->QueryInterface(IID_IDeckLinkVideoBuffer, (void**)&buf) != S_OK) return;
+        buf->StartAccess(bmdBufferAccessWrite);
+        void* ptr = nullptr;
+        buf->GetBytes(&ptr);
+        if (ptr) {
+            const uint32_t dst_stride  = static_cast<uint32_t>(vf->GetRowBytes());
+            const uint32_t copy_stride = std::min(stride, dst_stride);
+            for (int y = 0; y < video_info.height; ++y)
+                std::memcpy(static_cast<uint8_t*>(ptr) + y * dst_stride,
+                            src + y * stride, copy_stride);
+        }
+        buf->EndAccess(bmdBufferAccessWrite);
+        buf->Release();
+
+        const int64_t t = scheduled_time;
+        scheduled_time += video_info.fps_den;
+        decklink_output->ScheduleVideoFrame(vf, t, video_info.fps_den, video_info.fps_num);
+    }
+
+    // Converts float32 planar → interleaved int32 and pushes to DeckLink audio buffer.
+    void submit_audio(const float* planar, int samples) {
+        if (!has_audio || samples <= 0) return;
+        const int    ch    = audio_info.channels;
+        const size_t total = static_cast<size_t>(ch) * static_cast<size_t>(samples);
+        if (audio_convert_buf.size() < total) audio_convert_buf.resize(total);
+        auto& interleaved = audio_convert_buf;
+        for (int s = 0; s < samples; ++s)
+            for (int c = 0; c < ch; ++c) {
+                float v = std::max(-1.0f, std::min(1.0f, planar[c * samples + s]));
+                interleaved[static_cast<size_t>(s) * ch + c] =
+                    static_cast<int32_t>(v * 2147483647.0f);
+            }
+        const int64_t t = audio_stream_time;
+        audio_stream_time += samples;
+        uint32_t written = 0;
+        decklink_output->ScheduleAudioSamples(
+            interleaved.data(), static_cast<uint32_t>(samples),
+            t, audio_info.sample_rate, &written);
+    }
+
+private:
+    class OutputCallback : public IDeckLinkVideoOutputCallback {
+    public:
+        explicit OutputCallback(DeckLinkSender& owner) : owner(owner) {}
+
+        HRESULT ScheduledFrameCompleted(IDeckLinkVideoFrame* completed,
+                                        BMDOutputFrameCompletionResult) override {
+            {
+                std::lock_guard<std::mutex> lk(owner.pool_mutex);
+                owner.free_frames.push_back(
+                    static_cast<IDeckLinkMutableVideoFrame*>(completed));
+            }
+            owner.pool_cv.notify_one();
+            return S_OK;
+        }
+
+        HRESULT ScheduledPlaybackHasStopped() override { 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:
+        DeckLinkSender&    owner;
+        std::atomic<int32_t> ref_count{1};
+    };
+
+    // DeckLink SDK objects
+    std::vector<IDeckLink*>              raw_devices;
+    IDeckLink*                           selected_device = nullptr;
+    IDeckLinkOutput*                     decklink_output = nullptr;
+    OutputCallback*                      output_callback = nullptr;
+    std::vector<IDeckLinkMutableVideoFrame*> all_frames; // for destructor cleanup
+
+    // Frame pool — frames returned by ScheduledFrameCompleted land here
+    std::mutex                           pool_mutex;
+    std::condition_variable              pool_cv;
+    std::vector<IDeckLinkMutableVideoFrame*> free_frames;
+
+    // Scheduling state — only written from submit_frame/submit_audio (single-threaded caller)
+    int64_t              scheduled_time    = 0; // next video frame position (fps_num units)
+    int64_t              audio_stream_time = 0; // next audio batch position (sample units)
+    std::vector<int32_t> audio_convert_buf;     // reused across submit_audio calls
+
+    // Maps width/height/fps to a BMDDisplayMode. Uses float comparison to handle
+    // any representation of drop-frame rates (e.g. 30000/1001 or 2997/100).
+    static BMDDisplayMode pick_display_mode(int width, int height, int fps_num, int fps_den) {
+        const double fps = static_cast<double>(fps_num) / static_cast<double>(fps_den);
+        if (width == 1920 && height == 1080) {
+            if      (std::abs(fps - 23.976) < 0.01) return bmdModeHD1080p2398;
+            else if (std::abs(fps - 24.0)   < 0.01) return bmdModeHD1080p24;
+            else if (std::abs(fps - 25.0)   < 0.01) return bmdModeHD1080p25;
+            else if (std::abs(fps - 29.97)  < 0.01) return bmdModeHD1080p2997;
+            else if (std::abs(fps - 30.0)   < 0.01) return bmdModeHD1080p30;
+            else if (std::abs(fps - 50.0)   < 0.01) return bmdModeHD1080p50;
+            else if (std::abs(fps - 59.94)  < 0.01) return bmdModeHD1080p5994;
+            else if (std::abs(fps - 60.0)   < 0.01) return bmdModeHD1080p6000;
+        } else if (width == 1280 && height == 720) {
+            if      (std::abs(fps - 50.0)   < 0.01) return bmdModeHD720p50;
+            else if (std::abs(fps - 59.94)  < 0.01) return bmdModeHD720p5994;
+            else if (std::abs(fps - 60.0)   < 0.01) return bmdModeHD720p60;
+        }
+        throw std::runtime_error(
+            "No DeckLink display mode for " + std::to_string(width) + "x" +
+            std::to_string(height) + " @ " + std::to_string(fps_num) +
+            "/" + std::to_string(fps_den) + " fps");
+    }
+
+    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) {
+            IDeckLinkOutput* out = nullptr;
+            if (device->QueryInterface(IID_IDeckLinkOutput, (void**)&out) == S_OK) {
+                out->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 output devices found");
+    }
+};
+
+} // namespace dmf

shared/NDIReceiver.hpp

@@ -126,10 +126,13 @@ public:
     // 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) {
+                      std::vector<float>& audio_out, AudioInfo& audio_info,
+                      bool want_audio = false) {
         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);
+        auto type = NDIlib_recv_capture_v3(recv_, &video_frame,
+                                           want_audio ? &audio_frame : nullptr,
+                                           nullptr, 5);
 
         if (type == NDIlib_frame_type_error)
             throw std::runtime_error("NDI source lost");

shared/NodeBase.hpp

@@ -37,8 +37,6 @@ inline const char* mxl_status_str(mxlStatus s) noexcept {
     }
 }
 
-
-
 // Base class for all DMF node binaries.
 //
 // Handles the boilerplate every node needs:

shared/V210.hpp

@@ -60,29 +60,6 @@ inline void pack_block(
     w[3] = (y4     & 0x3FFu)        | ((p45.cr & 0x3FFu) << 10) | ((y5     & 0x3FFu) << 20);
 }
 
-// Write one horizontal line of bars.
-// `stride` is the line size in bytes as returned by MXL (configInfo.discrete.sliceSizes[0]).
-// Bytes beyond the active pixels are already zeroed by the mmap, so no explicit padding needed.
-inline void write_bar_line(uint8_t* line, int width, uint32_t /*stride*/)
-{
-    const int n = static_cast<int>(SMPTE_BARS.size());
-    const int blocks = width / 6;  // one V210 block = 6 pixels = 16 bytes
-
-    for (int b = 0; b < blocks; b++) {
-        int x = b * 6;
-        auto color = [&](int px) -> const Color& {
-            return SMPTE_BARS[static_cast<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);
-    }
-}
-
 // 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)
@@ -140,12 +117,6 @@ 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;
@@ -160,42 +131,39 @@ inline void UYVYtoV210(uint8_t* src_buf, uint8_t* dst_buf, int width, int height
             // 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, static_cast<uint16_t>(mp[0]<<2), static_cast<uint16_t>(mp[2]<<2)},
-                {0, (uint16_t)(mp[4]<<2), (uint16_t)(mp[6]<<2)}, (uint16_t)(mp[5]<<2), (uint16_t)(mp[7]<<2),
+                static_cast<uint16_t>(mp[1]<<2), static_cast<uint16_t>(mp[3]<<2),
-                {0, (uint16_t)(mp[8]<<2), (uint16_t)(mp[10]<<2)}, (uint16_t)(mp[9]<<2), (uint16_t)(mp[11]<<2)
+                {0, static_cast<uint16_t>(mp[4]<<2), static_cast<uint16_t>(mp[6]<<2)},
-            );
+                static_cast<uint16_t>(mp[5]<<2), static_cast<uint16_t>(mp[7]<<2),
+                {0, static_cast<uint16_t>(mp[8]<<2), static_cast<uint16_t>(mp[10]<<2)},
+                static_cast<uint16_t>(mp[9]<<2), static_cast<uint16_t>(mp[11]<<2));
         }
         src += src_stride;
         dst += dst_stride;
     }
 }
 
-inline void YUV422P10toV210(const uint16_t* y, const uint16_t* u, const uint16_t* v,
+inline void YUV422P10toV210(
+    const uint16_t* y, const uint16_t* u, const uint16_t* v,
     uint8_t* dst, int width, int height,
-    int y_stride, int u_stride, int v_stride,  // bytes between rows
+    int y_stride, int u_stride, int v_stride,
     uint32_t dst_stride)
-    {
+{
     for (int row = 0; row < height; row++) {
         const uint16_t* y_row = reinterpret_cast<const uint16_t*>(
-                reinterpret_cast<const uint8_t*>(y) + row * y_stride
+            reinterpret_cast<const uint8_t*>(y) + row * y_stride);
-            );
         const uint16_t* u_row = reinterpret_cast<const uint16_t*>(
-                reinterpret_cast<const uint8_t*>(u) + row * u_stride
+            reinterpret_cast<const uint8_t*>(u) + row * u_stride);
-            );
         const uint16_t* v_row = reinterpret_cast<const uint16_t*>(
-                reinterpret_cast<const uint8_t*>(v) + row * v_stride
+            reinterpret_cast<const uint8_t*>(v) + row * v_stride);
-            );
-
         uint8_t*        dst_row = dst + static_cast<ptrdiff_t>(row) * dst_stride;
         const int blocks = width / 6;
-
         for (int b = 0; b < blocks; b++) {
-                int x = b * 6;
+            const int      x   = b * 6;
             const uint16_t cb0 = u_row[x/2], cb1 = u_row[x/2+1], cb2 = u_row[x/2+2];
             const uint16_t cr0 = v_row[x/2], cr1 = v_row[x/2+1], cr2 = v_row[x/2+2];
             const uint16_t y0  = y_row[x],   y1  = y_row[x+1],   y2  = y_row[x+2];
             const uint16_t y3  = y_row[x+3], y4  = y_row[x+4],   y5  = y_row[x+5];
-
             auto* w = reinterpret_cast<uint32_t*>(dst_row + b * 16);
             w[0] = (cb0 & 0x3FFu) | ((y0 & 0x3FFu) << 10) | ((cr0 & 0x3FFu) << 20);
             w[1] = (y1  & 0x3FFu) | ((cb1 & 0x3FFu) << 10) | ((y2  & 0x3FFu) << 20);
@@ -203,6 +171,6 @@ inline void YUV422P10toV210(const uint16_t* y, const uint16_t* u, const uint16_t
             w[3] = (y4  & 0x3FFu) | ((cr2 & 0x3FFu) << 10) | ((y5  & 0x3FFu) << 20);
         }
     }
-    }
+}
 
 } // namespace dmf::v210