From 041588b990d835ac6d365adcfd19dc9179ed791f Mon Sep 17 00:00:00 2001 From: JohannesItten Date: Thu, 9 Jul 2026 18:31:53 +0300 Subject: [PATCH] pip based --- CMakeLists.txt | 1 + combiner.md | 138 ----------------------------- nodes/pip/CMakeLists.txt | 4 + nodes/pip/main.cpp | 184 +++++++++++++++++++++++++++++++++++++++ shared/V210.hpp | 102 ++++++++++++++++++++++ 5 files changed, 291 insertions(+), 138 deletions(-) delete mode 100644 combiner.md create mode 100644 nodes/pip/CMakeLists.txt create mode 100644 nodes/pip/main.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 5a725a1..7fc7eb0 100644 --- a/CMakeLists.txt +++ b/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 diff --git a/combiner.md b/combiner.md deleted file mode 100644 index c2bdca9..0000000 --- a/combiner.md +++ /dev/null @@ -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( - static_cast(in_slice.base.fragments[0].pointer) + c * in_slice.stride); - auto* dst0 = reinterpret_cast( - static_cast(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( - static_cast(in_slice.base.fragments[1].pointer) + c * in_slice.stride); - auto* dst1 = reinterpret_cast( - static_cast(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` diff --git a/nodes/pip/CMakeLists.txt b/nodes/pip/CMakeLists.txt new file mode 100644 index 0000000..85869ac --- /dev/null +++ b/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) diff --git a/nodes/pip/main.cpp b/nodes/pip/main.cpp new file mode 100644 index 0000000..abb50c3 --- /dev/null +++ b/nodes/pip/main.cpp @@ -0,0 +1,184 @@ +#include +#include +#include +#include +#include +#include +#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(); + const auto inset_id = config().at("inset_flow_id").at("id").get(); + const auto out_id = config().at("video_flow_id").at("id").get(); + + // 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(bg_h), out_cfg.discrete.grainCount); + + // --- Pre-allocate bilinear scaling workspace (reused every frame) --- + std::vector Y0(inset_w), Y1(inset_w); + std::vector Cb0(inset_w / 2), Cb1(inset_w / 2); + std::vector Cr0(inset_w / 2), Cr1(inset_w / 2); + + // --- Sync group: ensures both inputs have grain N before compositing --- + 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, timeout_count = 0; + + while (dmf::g_running.load(std::memory_order_relaxed)) { + mxlStatus st = mxlFlowSynchronizationGroupWaitForDataAt( + sync_group, index, 200'000'000); + + if (st == MXL_STATUS_OK) { + mxlGrainInfo bg_grain{}, inset_grain{}; + uint8_t* bg_buf = nullptr; + uint8_t* inset_buf = nullptr; + + mxlFlowReaderGetGrainNonBlocking(bg_reader, index, &bg_grain, &bg_buf); + mxlFlowReaderGetGrainNonBlocking(inset_reader, index, &inset_grain, &inset_buf); + + mxlGrainInfo out_grain{}; + uint8_t* out_buf = nullptr; + if (mxlFlowWriterOpenGrain(out_writer, index, &out_grain, &out_buf) == MXL_STATUS_OK) { + // Copy full background into output + std::memcpy(out_buf, bg_buf, + static_cast(bg_stride) * static_cast(bg_h)); + + // Scale inset and composite over background + if (inset_buf) { + 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++; + } + + index++; + + } else if (st == MXL_ERR_TIMEOUT) { + timeout_count++; + const uint64_t current = mxlGetCurrentIndex(&rate); + index = (current > index) ? current : index + 1; + if (timeout_count % 25 == 1) + log("sync timeout frames=%llu timeouts=%llu", frame_count, timeout_count); + + } else { + log("sync error (%s) at index=%llu", dmf::mxl_status_str(st), index); + break; + } + } + + log("stopped at index=%llu frames=%llu timeouts=%llu", + index, frame_count, timeout_count); + + mxlReleaseFlowSynchronizationGroup(instance(), sync_group); + mxlReleaseFlowReader(instance(), bg_reader); + mxlReleaseFlowReader(instance(), inset_reader); + mxlReleaseFlowWriter(instance(), out_writer); + } +}; + +int main() { + PiPNode node; + return node.execute(); +} diff --git a/shared/V210.hpp b/shared/V210.hpp index 391e1d0..241cdb9 100644 --- a/shared/V210.hpp +++ b/shared/V210.hpp @@ -1,8 +1,10 @@ #pragma once +#include #include #include #include #include +#include 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(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& Y0_buf, std::vector& Y1_buf, + std::vector& Cb0_buf, std::vector& Cb1_buf, + std::vector& Cr0_buf, std::vector& 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(inset_h) / pip_h; + const float inv_pip_w = static_cast(inset_w) / pip_w; + const float inv_pip_cw = static_cast(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(sy)); + const int sy1 = std::min(inset_h - 1, sy0 + 1); + const float fy = sy - static_cast(sy0); + + if (sy0 != cur_row0) { + unpack_row(inset + static_cast(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(sy1) * inset_stride, inset_w, + Y1_buf.data(), Cb1_buf.data(), Cr1_buf.data()); + cur_row1 = sy1; + } + + uint8_t* dst_row = dst + static_cast(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(sx)); + const int sx1 = std::min(inset_w - 1, sx0 + 1); + const float fx = sx - static_cast(sx0); + Y[i] = static_cast( + 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(cx)); + const int cx1 = std::min(inset_w / 2 - 1, cx0 + 1); + const float cfx = cx - static_cast(cx0); + Cb[i] = static_cast( + 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( + 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