itten/dmf-studio-rnd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source 
Reviewed-on: #6
This commit was merged in pull request #6.
This commit is contained in:
itten
2026-07-09 19:46:05 +03:00
parent 3820d0eeb7 c561bf569e
commit 30202e112e
5 changed files with 313 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CMakeLists.txt
+1
View File
@@ -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
-138
View File
@@ -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
+4
View File
@@ -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
+206
View File
@@ -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
+102
View File
@@ -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