Merge pull request 'Pip node' (#6) from pip-node into main

Reviewed-on: #6

CMakeLists.txt

@@ -113,6 +113,7 @@ if(DECKLINK_SDK_DIR)
 endif()
 
 add_subdirectory(nodes/videoin)
+add_subdirectory(nodes/pip)
 
 # ── Asio standalone (needed by Crow; no Boost) ───────────────────────────────
 FetchContent_Declare(asio_fc

combiner.md

@@ -1,138 +0,0 @@
-# AV Combiner Node — Implementation Steps
-
-Takes video from one upstream node and audio from another, outputs both as new MXL flows.
-Does NOT need a sync group — audio and video run at different rates and are handled independently.
-
----
-
-## Step 1 — Frontend (`dmf-studio-ui/src/nodeTypes.ts`)
-
-Add the node definition. Port IDs use `video_in-in` / `video_out-out` pattern so that
-`handleToPort()` produces distinct keys (`video_in_flow_id` vs `video_out_flow_id`).
-
-```typescript
-avcombiner: {
-  type: 'avcombiner',
-  label: 'AV Combiner',
-  ports: [
-    { id: 'video_in-in',   kind: 'video', direction: 'in'  },
-    { id: 'audio_in-in',   kind: 'audio', direction: 'in'  },
-    { id: 'video_out-out', kind: 'video', direction: 'out' },
-    { id: 'audio_out-out', kind: 'audio', direction: 'out' },
-  ],
-  params: [],
-},
-```
-
-Config keys the node receives:
-- `video_in_flow_id.id`  — input video flow UUID
-- `audio_in_flow_id.id`  — input audio flow UUID
-- `video_out_flow_id.id` — output video flow UUID
-- `audio_out_flow_id.id` — output audio flow UUID
-
----
-
-## Step 2 — CMake
-
-Create `nodes/avcombiner/CMakeLists.txt`:
-
-```cmake
-add_executable(dmf-node-avcombiner main.cpp)
-target_compile_features(dmf-node-avcombiner PRIVATE cxx_std_20)
-target_link_libraries(dmf-node-avcombiner PRIVATE dmf-shared)
-install(TARGETS dmf-node-avcombiner RUNTIME DESTINATION bin)
-```
-
-In the root `CMakeLists.txt`, add alongside the other nodes:
-
-```cmake
-add_subdirectory(nodes/avcombiner)
-```
-
----
-
-## Step 3 — `nodes/avcombiner/main.cpp`
-
-Structure (follow the same patterns as `ndiout`):
-
-```
-1. Check video_in_flow_id and video_out_flow_id both present — exit if not
-2. Check audio_in_flow_id and audio_out_flow_id (optional — audio is optional)
-
-3. Wait for video input flow active  (mxlIsFlowActive loop, 100ms sleep)
-4. Create video input reader         (mxlCreateFlowReader)
-5. Get video config info             (mxlFlowReaderGetConfigInfo → video_stride)
-6. Read video format from flow_def   (dmf::read_video_flow_info(domain(), flow_id))
-   → width, height, fps_num, fps_den
-
-7. If has_audio:
-   Wait for audio input flow active
-   Create audio input reader
-   Read audio format from flow_def   (dmf::read_audio_flow_info(domain(), flow_id))
-   → sample_rate, channels, samples_per_grain
-
-8. Create video output writer        (mxlCreateFlowWriter with make_video_flow_def)
-   → video_out_stride from configInfo.discrete.sliceSizes[0]
-
-9. If has_audio:
-   Create audio output writer        (mxlCreateFlowWriter with make_audio_flow_def)
-
-10. Init clocks:
-    video_index = mxlGetCurrentIndex(&video_rate)
-    audio_index = mxlGetCurrentIndex(&audio_rate)   // if has_audio
-
-11. Main loop (same pattern as ndiout):
-
-    // Audio — non-blocking
-    if (has_audio) {
-        mxlFlowReaderGetSamplesNonBlocking(audio_in_reader, audio_index, samples_per_grain, &in_slice)
-        if OK:
-            mxlFlowWriterOpenSamples(audio_out_writer, audio_index, samples_per_grain, &out_slice)
-            memcpy each channel fragment (frag0, frag1 wrap pattern)
-            mxlFlowWriterCommitSamples
-            audio_index += samples_per_grain
-        if TOO_LATE: jump to headIndex
-    }
-
-    // Video — non-blocking
-    mxlFlowReaderGetGrainNonBlocking(video_in_reader, video_index, &grain, &in_buf)
-    if OK:
-        mxlFlowWriterOpenGrain(video_out_writer, video_index, &out_grain, &out_buf)
-        memcpy(out_buf, in_buf, video_out_stride * height)
-        mxlFlowWriterCommitGrain
-        video_index++
-    if TOO_EARLY: mxlSleepForNs(1ms)
-    if TOO_LATE:  jump to headIndex
-```
-
-### Audio memcpy pattern (wrapped ring buffer)
-
-MXL audio slices can wrap around the ring buffer — always copy both fragments:
-
-```cpp
-const size_t frag0 = in_slice.base.fragments[0].size / sizeof(float);
-const size_t frag1 = in_slice.base.fragments[1].size / sizeof(float);
-for (int c = 0; c < channels; ++c) {
-    const auto* src0 = reinterpret_cast<const float*>(
-        static_cast<const uint8_t*>(in_slice.base.fragments[0].pointer) + c * in_slice.stride);
-    auto* dst0 = reinterpret_cast<float*>(
-        static_cast<uint8_t*>(out_slice.base.fragments[0].pointer) + c * out_slice.stride);
-    std::memcpy(dst0, src0, frag0 * sizeof(float));
-    if (frag1 > 0) {
-        const auto* src1 = reinterpret_cast<const float*>(
-            static_cast<const uint8_t*>(in_slice.base.fragments[1].pointer) + c * in_slice.stride);
-        auto* dst1 = reinterpret_cast<float*>(
-            static_cast<uint8_t*>(out_slice.base.fragments[1].pointer) + c * out_slice.stride);
-        std::memcpy(dst1, src1, frag1 * sizeof(float));
-    }
-}
-```
-
----
-
-## Reference nodes
-
-- Input flow setup (wait + reader + flow_def read): `nodes/ndiout/main.cpp`, `nodes/decklinkout/main.cpp`
-- Output flow setup (writer creation): `nodes/ndiin/main.cpp`, `nodes/decklinkin/main.cpp`
-- Audio memcpy pattern: `nodes/ndiout/main.cpp` lines 172–190
-- `make_video_flow_def` / `make_audio_flow_def` / `read_video_flow_info` / `read_audio_flow_info`: `shared/FlowDef.hpp`

nodes/pip/CMakeLists.txt

@@ -0,0 +1,4 @@
+add_executable(dmf-node-pip main.cpp)
+target_compile_features(dmf-node-pip PRIVATE cxx_std_20)
+target_link_libraries(dmf-node-pip PRIVATE dmf-shared)
+install(TARGETS dmf-node-pip RUNTIME DESTINATION bin)

nodes/pip/main.cpp

@@ -0,0 +1,206 @@
+#include <algorithm>
+#include <cstring>
+#include <string>
+#include <vector>
+#include <mxl/flow.h>
+#include <mxl/time.h>
+#include "NodeBase.hpp"
+#include "FlowDef.hpp"
+#include "V210.hpp"
+
+// Round down to nearest V210-aligned pixel count (multiple of 6).
+static int v210_align(int pixels) { return (pixels / 6) * 6; }
+
+class PiPNode : public dmf::NodeBase {
+    void run() override {
+        if (!config().contains("background_flow_id")) { log("no background connected"); return; }
+        if (!config().contains("inset_flow_id"))      { log("no inset connected");      return; }
+        if (!config().contains("video_flow_id"))      { log("no output connected");      return; }
+
+        const auto bg_id    = config().at("background_flow_id").at("id").get<std::string>();
+        const auto inset_id = config().at("inset_flow_id").at("id").get<std::string>();
+        const auto out_id   = config().at("video_flow_id").at("id").get<std::string>();
+
+        // Position and size of the inset in the output frame.
+        // x and width are snapped to 6-pixel V210 boundaries.
+        const int pip_x = v210_align(config().value("x", 0));
+        const int pip_y = config().value("y", 0);
+        const int pip_w = v210_align(config().value("width",  480));
+        const int pip_h = config().value("height", 270);
+
+        // --- Wait for both input flows ---
+        for (const auto* fid : {&bg_id, &inset_id}) {
+            log("waiting for flow %s...", fid->c_str());
+            bool active = false;
+            while (!active && dmf::g_running.load(std::memory_order_relaxed)) {
+                mxlIsFlowActive(instance(), fid->c_str(), &active);
+                if (!active) mxlSleepForNs(100'000'000);
+            }
+            if (!dmf::g_running) return;
+        }
+
+        // --- Create readers ---
+        mxlFlowReader     bg_reader{}, inset_reader{};
+        mxlFlowConfigInfo bg_cfg{}, inset_cfg{};
+
+        if (mxlCreateFlowReader(instance(), bg_id.c_str(), "", &bg_reader) != MXL_STATUS_OK) {
+            log("background mxlCreateFlowReader failed"); return;
+        }
+        if (mxlCreateFlowReader(instance(), inset_id.c_str(), "", &inset_reader) != MXL_STATUS_OK) {
+            log("inset mxlCreateFlowReader failed");
+            mxlReleaseFlowReader(instance(), bg_reader);
+            return;
+        }
+        mxlFlowReaderGetConfigInfo(bg_reader,    &bg_cfg);
+        mxlFlowReaderGetConfigInfo(inset_reader, &inset_cfg);
+
+        const uint32_t bg_stride    = bg_cfg.discrete.sliceSizes[0];
+        const uint32_t inset_stride = inset_cfg.discrete.sliceSizes[0];
+
+        // --- Read formats from flow_def.json ---
+        const auto bg_fi    = dmf::read_video_flow_info(domain(), bg_id);
+        const auto inset_fi = dmf::read_video_flow_info(domain(), inset_id);
+
+        const int bg_w    = bg_fi.width;
+        const int bg_h    = bg_fi.height;
+        const int fps_num = bg_fi.fps_num;
+        const int fps_den = bg_fi.fps_den;
+        const int inset_w = inset_fi.width;
+        const int inset_h = inset_fi.height;
+
+        log("background: %dx%d @ %d/%d fps  stride=%u",
+            bg_w, bg_h, fps_num, fps_den, bg_stride);
+        log("inset src:  %dx%d  stride=%u", inset_w, inset_h, inset_stride);
+        log("pip region: %dx%d at (%d,%d)", pip_w, pip_h, pip_x, pip_y);
+
+        // Clamp pip region to background bounds
+        const int clamped_w = v210_align(std::min(pip_w, bg_w - pip_x));
+        const int clamped_h = std::min(pip_h, bg_h - pip_y);
+        if (clamped_w <= 0 || clamped_h <= 0) {
+            log("pip region is outside background bounds — exiting");
+            mxlReleaseFlowReader(instance(), bg_reader);
+            mxlReleaseFlowReader(instance(), inset_reader);
+            return;
+        }
+
+        // --- Create output writer (same format as background) ---
+        mxlFlowWriter     out_writer{};
+        mxlFlowConfigInfo out_cfg{};
+        bool created = false;
+        mxlStatus vst = mxlCreateFlowWriter(
+            instance(),
+            dmf::make_video_flow_def(out_id, node_id(), bg_w, bg_h, fps_num, fps_den).c_str(),
+            "", &out_writer, &out_cfg, &created);
+        if (vst != MXL_STATUS_OK) {
+            log("mxlCreateFlowWriter failed (%s)", dmf::mxl_status_str(vst));
+            mxlReleaseFlowReader(instance(), bg_reader);
+            mxlReleaseFlowReader(instance(), inset_reader);
+            return;
+        }
+        const uint32_t out_stride = out_cfg.discrete.sliceSizes[0];
+        log("output: stride=%u  grain=%u B  ring=%u grains",
+            out_stride, out_stride * static_cast<uint32_t>(bg_h), out_cfg.discrete.grainCount);
+
+        // --- Pre-allocate bilinear scaling workspace (reused every frame) ---
+        std::vector<uint16_t> Y0(inset_w), Y1(inset_w);
+        std::vector<uint16_t> Cb0(inset_w / 2), Cb1(inset_w / 2);
+        std::vector<uint16_t> Cr0(inset_w / 2), Cr1(inset_w / 2);
+
+        // --- Sync group: blocks until both inputs have grain N ---
+        mxlFlowSynchronizationGroup sync_group{};
+        mxlCreateFlowSynchronizationGroup(instance(), &sync_group);
+        mxlFlowSynchronizationGroupAddReader(sync_group, bg_reader);
+        mxlFlowSynchronizationGroupAddReader(sync_group, inset_reader);
+
+        // --- Clock ---
+        const mxlRational rate = {fps_num, fps_den};
+        uint64_t index = mxlGetCurrentIndex(&rate);
+        log("start index=%llu", index);
+
+        uint64_t frame_count = 0, stall_count = 0;
+        bool fatal = false;
+
+        while (dmf::g_running.load(std::memory_order_relaxed)) {
+            // WaitForDataAt expects TAI nanoseconds, not a grain index.
+            const uint64_t tai_ns = mxlIndexToTimestamp(&rate, index);
+            const mxlStatus st = mxlFlowSynchronizationGroupWaitForDataAt(
+                sync_group, tai_ns, 200'000'000);
+
+            if (st == MXL_STATUS_OK) {
+                mxlGrainInfo bg_grain{}, inset_grain{};
+                uint8_t* bg_buf    = nullptr;
+                uint8_t* inset_buf = nullptr;
+
+                // Use blocking reads: sync group returns OK when headIndex >= expectedIndex,
+                // but the writer may not have set validSlices yet. Blocking read handles
+                // that race by waiting for validSlices == totalSlices.
+                const mxlStatus bg_st = mxlFlowReaderGetGrain(
+                    bg_reader,    index, 40'000'000, &bg_grain, &bg_buf);
+                const mxlStatus in_st = mxlFlowReaderGetGrain(
+                    inset_reader, index, 40'000'000, &inset_grain, &inset_buf);
+
+                if (bg_st != MXL_STATUS_OK || in_st != MXL_STATUS_OK || !bg_buf || !inset_buf) {
+                    log("grain read after sync OK: bg=%s inset=%s at index=%llu",
+                        dmf::mxl_status_str(bg_st), dmf::mxl_status_str(in_st), index);
+                    index++;
+                } else {
+                    mxlGrainInfo out_grain{};
+                    uint8_t*     out_buf = nullptr;
+                    const mxlStatus wst = mxlFlowWriterOpenGrain(
+                        out_writer, index, &out_grain, &out_buf);
+                    if (wst != MXL_STATUS_OK) {
+                        log("writer OpenGrain failed (%s) at index=%llu",
+                            dmf::mxl_status_str(wst), index);
+                    } else {
+                        std::memcpy(out_buf, bg_buf,
+                                    static_cast<size_t>(bg_stride) * static_cast<size_t>(bg_h));
+
+                        dmf::v210::scale_and_overlay(
+                            inset_buf, inset_stride, inset_w, inset_h,
+                            out_buf,   out_stride,
+                            pip_x, pip_y, clamped_w, clamped_h,
+                            Y0, Y1, Cb0, Cb1, Cr0, Cr1);
+
+                        out_grain.flags = (bg_grain.flags | inset_grain.flags) & MXL_GRAIN_FLAG_INVALID;
+                        out_grain.validSlices = out_grain.totalSlices;
+                        mxlFlowWriterCommitGrain(out_writer, &out_grain);
+                        frame_count++;
+                        if (frame_count % 25 == 0)
+                            log("heartbeat frames=%llu  stalls=%llu  index=%llu",
+                                frame_count, stall_count, index);
+                    }
+                    index++;
+                }
+
+            } else if (st == MXL_ERR_TIMEOUT ||
+                       st == MXL_ERR_OUT_OF_RANGE_TOO_EARLY ||
+                       st == MXL_ERR_OUT_OF_RANGE_TOO_LATE) {
+                stall_count++;
+                const uint64_t current = mxlGetCurrentIndex(&rate);
+                const uint64_t next = (current > index) ? current : index + 1;
+                log("sync stall (%s) at index=%llu → jumping to %llu  frames=%llu",
+                    dmf::mxl_status_str(st), index, next, frame_count);
+                index = next;
+
+            } else {
+                log("sync fatal (%s) at index=%llu", dmf::mxl_status_str(st), index);
+                fatal = true;
+                break;
+            }
+        }
+
+        log("stopped: %s  frames=%llu  stalls=%llu  index=%llu",
+            fatal ? "fatal error" : "shutdown signal",
+            frame_count, stall_count, index);
+
+        mxlReleaseFlowSynchronizationGroup(instance(), sync_group);
+        mxlReleaseFlowReader(instance(), bg_reader);
+        mxlReleaseFlowReader(instance(), inset_reader);
+        mxlReleaseFlowWriter(instance(), out_writer);
+    }
+};
+
+int main() {
+    PiPNode node;
+    return node.execute();
+}

shared/V210.hpp

@@ -1,8 +1,10 @@
 #pragma once
+#include <algorithm>
 #include <array>
 #include <cstddef>
 #include <cstdint>
 #include <cstring>
+#include <vector>
 
 namespace dmf::v210 {
 
@@ -173,4 +175,104 @@ inline void YUV422P10toV210(
     }
 }
 
+// Unpack one V210 row into planar uint16_t Y (width values),
+// Cb and Cr (width/2 values each). Width must be a multiple of 6.
+inline void unpack_row(const uint8_t* src, int width,
+                       uint16_t* Y, uint16_t* Cb, uint16_t* Cr)
+{
+    const auto* w = reinterpret_cast<const uint32_t*>(src);
+    const int blocks = width / 6;
+    for (int b = 0; b < blocks; ++b, w += 4) {
+        const int x = b * 6;
+        Cb[x/2]   = (w[0] >>  0) & 0x3FF;
+        Y[x]      = (w[0] >> 10) & 0x3FF;
+        Cr[x/2]   = (w[0] >> 20) & 0x3FF;
+        Y[x+1]    = (w[1] >>  0) & 0x3FF;
+        Cb[x/2+1] = (w[1] >> 10) & 0x3FF;
+        Y[x+2]    = (w[1] >> 20) & 0x3FF;
+        Cr[x/2+1] = (w[2] >>  0) & 0x3FF;
+        Y[x+3]    = (w[2] >> 10) & 0x3FF;
+        Cb[x/2+2] = (w[2] >> 20) & 0x3FF;
+        Y[x+4]    = (w[3] >>  0) & 0x3FF;
+        Cr[x/2+2] = (w[3] >> 10) & 0x3FF;
+        Y[x+5]    = (w[3] >> 20) & 0x3FF;
+    }
+}
+
+// Scale the inset V210 frame into a rectangular region of dst using bilinear
+// interpolation. pip_x and pip_w must be multiples of 6 (V210 alignment).
+// Workspace vectors are passed in to avoid per-call heap allocation.
+inline void scale_and_overlay(
+    const uint8_t* inset,  uint32_t inset_stride, int inset_w, int inset_h,
+    uint8_t*       dst,    uint32_t dst_stride,
+    int pip_x, int pip_y, int pip_w, int pip_h,
+    std::vector<uint16_t>& Y0_buf,  std::vector<uint16_t>& Y1_buf,
+    std::vector<uint16_t>& Cb0_buf, std::vector<uint16_t>& Cb1_buf,
+    std::vector<uint16_t>& Cr0_buf, std::vector<uint16_t>& Cr1_buf)
+{
+    Y0_buf.resize(inset_w);  Y1_buf.resize(inset_w);
+    Cb0_buf.resize(inset_w / 2); Cb1_buf.resize(inset_w / 2);
+    Cr0_buf.resize(inset_w / 2); Cr1_buf.resize(inset_w / 2);
+
+    const int out_blocks   = pip_w / 6;
+    const int dst_x_bytes  = (pip_x / 6) * 16;
+    const float inv_pip_h  = static_cast<float>(inset_h) / pip_h;
+    const float inv_pip_w  = static_cast<float>(inset_w) / pip_w;
+    const float inv_pip_cw = static_cast<float>(inset_w / 2) / (pip_w / 2);
+
+    int cur_row0 = -1, cur_row1 = -1;
+
+    for (int dy = 0; dy < pip_h; ++dy) {
+        const float sy  = (dy + 0.5f) * inv_pip_h - 0.5f;
+        const int   sy0 = std::max(0, static_cast<int>(sy));
+        const int   sy1 = std::min(inset_h - 1, sy0 + 1);
+        const float fy  = sy - static_cast<float>(sy0);
+
+        if (sy0 != cur_row0) {
+            unpack_row(inset + static_cast<size_t>(sy0) * inset_stride, inset_w,
+                       Y0_buf.data(), Cb0_buf.data(), Cr0_buf.data());
+            cur_row0 = sy0;
+        }
+        if (sy1 != cur_row1) {
+            unpack_row(inset + static_cast<size_t>(sy1) * inset_stride, inset_w,
+                       Y1_buf.data(), Cb1_buf.data(), Cr1_buf.data());
+            cur_row1 = sy1;
+        }
+
+        uint8_t* dst_row = dst + static_cast<size_t>(pip_y + dy) * dst_stride + dst_x_bytes;
+
+        for (int b = 0; b < out_blocks; ++b) {
+            const int bx = b * 6;
+            uint16_t Y[6], Cb[3], Cr[3];
+
+            for (int i = 0; i < 6; ++i) {
+                const float sx  = (bx + i + 0.5f) * inv_pip_w - 0.5f;
+                const int   sx0 = std::max(0, static_cast<int>(sx));
+                const int   sx1 = std::min(inset_w - 1, sx0 + 1);
+                const float fx  = sx - static_cast<float>(sx0);
+                Y[i] = static_cast<uint16_t>(
+                    Y0_buf[sx0] * (1-fx) * (1-fy) + Y0_buf[sx1] * fx * (1-fy) +
+                    Y1_buf[sx0] * (1-fx) *    fy  + Y1_buf[sx1] * fx *    fy  + 0.5f);
+            }
+            for (int i = 0; i < 3; ++i) {
+                const float cx  = (b * 3 + i + 0.5f) * inv_pip_cw - 0.5f;
+                const int   cx0 = std::max(0, static_cast<int>(cx));
+                const int   cx1 = std::min(inset_w / 2 - 1, cx0 + 1);
+                const float cfx = cx - static_cast<float>(cx0);
+                Cb[i] = static_cast<uint16_t>(
+                    Cb0_buf[cx0]*(1-cfx)*(1-fy) + Cb0_buf[cx1]*cfx*(1-fy) +
+                    Cb1_buf[cx0]*(1-cfx)*   fy  + Cb1_buf[cx1]*cfx*   fy  + 0.5f);
+                Cr[i] = static_cast<uint16_t>(
+                    Cr0_buf[cx0]*(1-cfx)*(1-fy) + Cr0_buf[cx1]*cfx*(1-fy) +
+                    Cr1_buf[cx0]*(1-cfx)*   fy  + Cr1_buf[cx1]*cfx*   fy  + 0.5f);
+            }
+
+            pack_block(dst_row + b * 16,
+                {0, Cb[0], Cr[0]}, Y[0], Y[1],
+                {0, Cb[1], Cr[1]}, Y[2], Y[3],
+                {0, Cb[2], Cr[2]}, Y[4], Y[5]);
+        }
+    }
+}
+
 } // namespace dmf::v210