st-2110-20 pipeline API MTL node in

.vscode/c_cpp_properties.json

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

CMakeLists.txt

@@ -76,6 +76,35 @@ FetchContent_MakeAvailable(json)
 # ── FFmpeg ──────────────────────────────────────────────────────────────────
 find_package(PkgConfig REQUIRED)
 
+# ── Media Transport Library (SMPTE ST 2110) ─────────────────────────────────
+# MTL is Meson-based, so consume an installed Meson build instead of adding the
+# source tree as a CMake subdirectory.
+set(MTL_SDK_DIR "" CACHE PATH "Path to installed Media Transport Library prefix")
+if(MTL_SDK_DIR)
+    find_library(MTL_LIBRARY
+        NAMES mtl
+        PATHS "${MTL_SDK_DIR}/lib" "${MTL_SDK_DIR}/lib64"
+        NO_DEFAULT_PATH
+        REQUIRED
+    )
+
+    add_library(mtl::mtl SHARED IMPORTED GLOBAL)
+    set_target_properties(mtl::mtl PROPERTIES
+        IMPORTED_LOCATION "${MTL_LIBRARY}"
+        INTERFACE_INCLUDE_DIRECTORIES "${MTL_SDK_DIR}/include"
+    )
+else()
+    pkg_check_modules(MTL IMPORTED_TARGET mtl)
+    if(MTL_FOUND)
+        add_library(mtl::mtl INTERFACE IMPORTED GLOBAL)
+        target_link_libraries(mtl::mtl INTERFACE PkgConfig::MTL)
+        target_include_directories(mtl::mtl INTERFACE ${MTL_INCLUDE_DIRS})
+        target_link_directories(mtl::mtl INTERFACE ${MTL_LIBRARY_DIRS})
+        target_compile_options(mtl::mtl INTERFACE ${MTL_CFLAGS_OTHER})
+        target_link_options(mtl::mtl INTERFACE ${MTL_LDFLAGS_OTHER})
+    endif()
+endif()
+
 # Check for FFmpeg components
 pkg_check_modules(FFMPEG REQUIRED IMPORTED_TARGET
     libavformat
@@ -112,6 +141,14 @@ if(DECKLINK_SDK_DIR)
     add_subdirectory(nodes/decklinkout)
 endif()
 
+# ── SMPTE-2110 (Media Transport Library) nodes ──────────────────────────────
+if(TARGET mtl::mtl)
+    add_subdirectory(nodes/st2110in)
+else()
+    message(STATUS "Media Transport Library not found; skipping SMPTE ST 2110 nodes")
+endif()
+
+
 add_subdirectory(nodes/videoin)
 add_subdirectory(nodes/pip)
 add_subdirectory(nodes/gaindb)

flow-graphs/2110-20.json

@@ -0,0 +1,28 @@
+{
+  "nodes": [
+    {
+      "id": "2110in",
+      "type": "2110",
+      "params": {
+        "interface": "eno1np0",
+        "local_ip": "192.168.0.3",
+        "source_ip": "192.168.0.2",
+        "mcast_ip": "239.255.197.181",
+        "udp_port": 16388,
+        "payload_type": 96,
+        "width": 1920,
+        "height": 1080,
+        "fps_num": 25,
+        "fps_den": 1
+      }
+    },
+    { "id": "fakesink",      "type": "fakesink",      "params": {} }
+  ],
+  "edges": [
+    {
+      "from": "2110in", "from_port": "video_flow_id",
+      "to":   "fakesink",      "to_port":   "video_flow_id",
+      "format": { "kind": "video", "width": 1920, "height": 1080, "fps_num": 25, "fps_den": 1 }
+    }
+  ]
+}

nodes/st2110in/CMakeLists.txt

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

nodes/st2110in/main.cpp

@@ -0,0 +1,91 @@
+#include <atomic>
+#include <cstdint>
+#include <exception>
+
+#include <mxl/flow.h>
+#include <mxl/time.h>
+
+#include "NodeBase.hpp"
+#include "ST2110Receiver.hpp"
+#include "V210.hpp"
+
+namespace dmf {
+
+class ST2110In : public NodeBase {
+    void run() override {
+        ST2110ReceiverConfig cfg;
+        try {
+            cfg = parse_st2110_receiver_config(config());
+        } catch (const std::exception& e) {
+            log("config error: %s", e.what());
+            return;
+        }
+
+        log("SMPTE 2110-20 RX %s:%u from %s on %s local=%s %dx%d @ %d/%d",
+            cfg.mcast_ip.c_str(), cfg.udp_port, cfg.source_ip.c_str(), cfg.ifname.c_str(),
+            cfg.local_ip.c_str(), cfg.width, cfg.height, cfg.fps_num, cfg.fps_den);
+
+        try {
+            MTLContext mtl(cfg);
+            ST20RxSession rx(mtl.get(), mtl.port_name(), cfg);
+            MXLVideoWriter writer(instance(), cfg, node_id());
+
+            const mxlRational video_rate = {cfg.fps_num, cfg.fps_den};
+            const uint32_t video_stride = writer.config().discrete.sliceSizes[0];
+            const uint32_t uyvy_stride = static_cast<uint32_t>(cfg.width * 2);
+            uint64_t frames_written = 0;
+
+            while (g_running.load(std::memory_order_relaxed)) {
+                st_frame* frame = st20p_rx_get_frame(rx.get());
+                if (!frame) {
+                    continue;
+                }
+
+                if (!st_is_frame_complete(frame->status)) {
+                    st20p_rx_put_frame(rx.get(), frame);
+                    continue;
+                }
+
+                if (frame->fmt != ST_FRAME_FMT_UYVY) {
+                    log("unexpected MTL frame fmt=%d; expected UYVY", frame->fmt);
+                    st20p_rx_put_frame(rx.get(), frame);
+                    continue;
+                }
+
+                mxlGrainInfo grain{};
+                uint8_t* video_buf = nullptr;
+                const uint64_t video_index = mxlGetCurrentIndex(&video_rate);
+                mxlStatus st = mxlFlowWriterOpenGrain(writer.get(), video_index, &grain, &video_buf);
+                if (st != MXL_STATUS_OK) {
+                    log("mxlFlowWriterOpenGrain failed (%s) index=%llu", mxl_status_str(st),
+                        static_cast<unsigned long long>(video_index));
+                    st20p_rx_put_frame(rx.get(), frame);
+                    continue;
+                }
+
+                v210::UYVYtoV210(static_cast<const uint8_t*>(frame->addr[0]), video_buf,
+                                 cfg.width, cfg.height, uyvy_stride, video_stride);
+                grain.flags = 0;
+                grain.validSlices = grain.totalSlices;
+                mxlFlowWriterCommitGrain(writer.get(), &grain);
+                st20p_rx_put_frame(rx.get(), frame);
+
+                frames_written++;
+                if (frames_written % 100 == 0) {
+                    log("received %llu frames", static_cast<unsigned long long>(frames_written));
+                }
+            }
+
+            log("stopped after %llu frames", static_cast<unsigned long long>(frames_written));
+        } catch (const std::exception& e) {
+            log("error: %s", e.what());
+        }
+    }
+};
+
+} // namespace dmf
+
+int main() {
+    dmf::ST2110In node;
+    return node.execute();
+}

shared/ST2110Receiver.hpp

@@ -0,0 +1,240 @@
+#pragma once
+
+#include <arpa/inet.h>
+
+#include <cstdint>
+#include <cstdio>
+#include <stdexcept>
+#include <string>
+
+#include <nlohmann/json.hpp>
+
+extern "C" {
+#include <mtl/mtl_api.h>
+#include <mtl/st20_api.h>
+#include <mtl/st_pipeline_api.h>
+}
+
+#include <mxl/flow.h>
+
+#include "FlowDef.hpp"
+#include "NodeBase.hpp"
+
+namespace dmf {
+
+struct ST2110ReceiverConfig {
+    std::string flow_id;
+    std::string ifname;
+    std::string local_ip;
+    std::string source_ip;
+    std::string mcast_ip;
+    uint16_t udp_port = 0;
+    uint8_t payload_type = 96;
+    int width = 1920;
+    int height = 1080;
+    int fps_num = 25;
+    int fps_den = 1;
+    int framebuff_cnt = 3;
+};
+
+inline void st2110_set_ip(uint8_t dst[MTL_IP_ADDR_LEN], const std::string& ip) {
+    if (inet_pton(AF_INET, ip.c_str(), dst) != 1) {
+        throw std::runtime_error("invalid IP address: " + ip);
+    }
+}
+
+inline uint16_t st2110_checked_u16(const nlohmann::json& j, const char* key) {
+    const int value = j.at(key).get<int>();
+    if (value < 0 || value > 65535) {
+        throw std::runtime_error(std::string(key) + " out of uint16 range");
+    }
+    return static_cast<uint16_t>(value);
+}
+
+inline uint8_t st2110_checked_u8(const nlohmann::json& j, const char* key, int fallback) {
+    const int value = j.value(key, fallback);
+    if (value < 0 || value > 255) {
+        throw std::runtime_error(std::string(key) + " out of uint8 range");
+    }
+    return static_cast<uint8_t>(value);
+}
+
+inline enum st_fps st2110_to_st_fps(int fps_num, int fps_den) {
+    if (fps_den == 1) {
+        switch (fps_num) {
+            case 24: return ST_FPS_P24;
+            case 25: return ST_FPS_P25;
+            case 30: return ST_FPS_P30;
+            case 50: return ST_FPS_P50;
+            case 60: return ST_FPS_P60;
+            case 100: return ST_FPS_P100;
+            case 120: return ST_FPS_P120;
+            default: break;
+        }
+    }
+    if (fps_num == 24000 && fps_den == 1001) return ST_FPS_P23_98;
+    if (fps_num == 30000 && fps_den == 1001) return ST_FPS_P29_97;
+    if (fps_num == 60000 && fps_den == 1001) return ST_FPS_P59_94;
+    if (fps_num == 120000 && fps_den == 1001) return ST_FPS_P119_88;
+    throw std::runtime_error("unsupported ST 2110 frame rate");
+}
+
+inline ST2110ReceiverConfig parse_st2110_receiver_config(const nlohmann::json& cfg) {
+    if (!cfg.contains("video_flow_id")) {
+        throw std::runtime_error("no video output connected");
+    }
+
+    ST2110ReceiverConfig out;
+    out.flow_id = cfg.at("video_flow_id").at("id").get<std::string>();
+    out.ifname = cfg.at("interface").get<std::string>();
+    out.local_ip = cfg.at("local_ip").get<std::string>();
+    out.source_ip = cfg.at("source_ip").get<std::string>();
+    out.mcast_ip = cfg.at("mcast_ip").get<std::string>();
+    out.udp_port = st2110_checked_u16(cfg, "udp_port");
+    out.payload_type = st2110_checked_u8(cfg, "payload_type", 96);
+    out.width = cfg.value("width", out.width);
+    out.height = cfg.value("height", out.height);
+    out.fps_num = cfg.value("fps_num", out.fps_num);
+    out.fps_den = cfg.value("fps_den", out.fps_den);
+    out.framebuff_cnt = cfg.value("framebuff_cnt", out.framebuff_cnt);
+
+    if (out.width <= 0 || out.height <= 0) {
+        throw std::runtime_error("width and height must be positive");
+    }
+    if (out.width % 6 != 0) {
+        throw std::runtime_error("width must be divisible by 6 for v210 output");
+    }
+    if (out.fps_num <= 0 || out.fps_den <= 0) {
+        throw std::runtime_error("fps_num and fps_den must be positive");
+    }
+    if (out.framebuff_cnt < 2 || out.framebuff_cnt > ST20_FB_MAX_COUNT) {
+        throw std::runtime_error("framebuff_cnt must be in [2, ST20_FB_MAX_COUNT]");
+    }
+
+    return out;
+}
+
+class MTLContext {
+public:
+    explicit MTLContext(const ST2110ReceiverConfig& cfg) {
+        mtl_init_params params{};
+        params.num_ports = 1;
+
+        port_name_ = "kernel:" + cfg.ifname;
+        std::snprintf(params.port[MTL_PORT_P], sizeof(params.port[MTL_PORT_P]), "%s",
+                      port_name_.c_str());
+
+        params.pmd[MTL_PORT_P] = MTL_PMD_KERNEL_SOCKET;
+        params.net_proto[MTL_PORT_P] = MTL_PROTO_STATIC;
+        params.rx_queues_cnt[MTL_PORT_P] = 1;
+        params.tx_queues_cnt[MTL_PORT_P] = 0;
+        params.log_level = MTL_LOG_LEVEL_INFO;
+        params.flags = MTL_FLAG_DEV_AUTO_START_STOP;
+        st2110_set_ip(params.sip_addr[MTL_PORT_P], cfg.local_ip);
+
+        handle_ = mtl_init(&params);
+        if (!handle_) {
+            throw std::runtime_error("mtl_init failed");
+        }
+    }
+
+    ~MTLContext() {
+        if (handle_) {
+            mtl_uninit(handle_);
+        }
+    }
+
+    MTLContext(const MTLContext&) = delete;
+    MTLContext& operator=(const MTLContext&) = delete;
+
+    mtl_handle get() const { return handle_; }
+    const std::string& port_name() const { return port_name_; }
+
+private:
+    mtl_handle handle_{nullptr};
+    std::string port_name_;
+};
+
+class ST20RxSession {
+public:
+    ST20RxSession(mtl_handle mt, const std::string& port_name,
+                  const ST2110ReceiverConfig& cfg) {
+        st20p_rx_ops ops{};
+        ops.name = "dmf-st2110in-video";
+        ops.port.num_port = 1;
+        ops.port.udp_port[MTL_SESSION_PORT_P] = cfg.udp_port;
+        ops.port.payload_type = cfg.payload_type;
+        std::snprintf(ops.port.port[MTL_SESSION_PORT_P],
+                      sizeof(ops.port.port[MTL_SESSION_PORT_P]), "%s", port_name.c_str());
+        st2110_set_ip(ops.port.ip_addr[MTL_SESSION_PORT_P], cfg.mcast_ip);
+        st2110_set_ip(ops.port.mcast_sip_addr[MTL_SESSION_PORT_P], cfg.source_ip);
+
+        ops.width = static_cast<uint32_t>(cfg.width);
+        ops.height = static_cast<uint32_t>(cfg.height);
+        ops.fps = st2110_to_st_fps(cfg.fps_num, cfg.fps_den);
+        ops.interlaced = false;
+        ops.transport_fmt = ST20_FMT_YUV_422_8BIT;
+        ops.output_fmt = ST_FRAME_FMT_UYVY;
+        ops.device = ST_PLUGIN_DEVICE_AUTO;
+        ops.framebuff_cnt = static_cast<uint16_t>(cfg.framebuff_cnt);
+        ops.flags = ST20P_RX_FLAG_BLOCK_GET;
+
+        handle_ = st20p_rx_create(mt, &ops);
+        if (!handle_) {
+            throw std::runtime_error("st20p_rx_create failed");
+        }
+        st20p_rx_set_block_timeout(handle_, 100'000'000);
+    }
+
+    ~ST20RxSession() {
+        if (handle_) {
+            st20p_rx_wake_block(handle_);
+            st20p_rx_free(handle_);
+        }
+    }
+
+    ST20RxSession(const ST20RxSession&) = delete;
+    ST20RxSession& operator=(const ST20RxSession&) = delete;
+
+    st20p_rx_handle get() const { return handle_; }
+
+private:
+    st20p_rx_handle handle_{nullptr};
+};
+
+class MXLVideoWriter {
+public:
+    MXLVideoWriter(mxlInstance instance, const ST2110ReceiverConfig& cfg,
+                   const std::string& node_id)
+        : instance_(instance) {
+        bool created = false;
+        const std::string flow_def =
+            make_video_flow_def(cfg.flow_id, node_id, cfg.width, cfg.height, cfg.fps_num,
+                                cfg.fps_den);
+        const mxlStatus st =
+            mxlCreateFlowWriter(instance_, flow_def.c_str(), "", &writer_, &config_, &created);
+        if (st != MXL_STATUS_OK) {
+            throw std::runtime_error(std::string("mxlCreateFlowWriter failed: ") +
+                                     mxl_status_str(st));
+        }
+    }
+
+    ~MXLVideoWriter() {
+        if (writer_) {
+            mxlReleaseFlowWriter(instance_, writer_);
+        }
+    }
+
+    MXLVideoWriter(const MXLVideoWriter&) = delete;
+    MXLVideoWriter& operator=(const MXLVideoWriter&) = delete;
+
+    mxlFlowWriter get() const { return writer_; }
+    const mxlFlowConfigInfo& config() const { return config_; }
+
+private:
+    mxlInstance instance_{nullptr};
+    mxlFlowWriter writer_{nullptr};
+    mxlFlowConfigInfo config_{};
+};
+
+} // namespace dmf

shared/V210.hpp

@@ -119,7 +119,7 @@ inline void fill_white(uint8_t* buf, int width, int height, uint32_t stride)
     fill_solid(buf, width, height, stride, {940, 512, 512});
 }
 
-inline void UYVYtoV210(uint8_t* src_buf, uint8_t* dst_buf, int width, int height, uint32_t src_stride, uint32_t dst_stride)
+inline void UYVYtoV210(const 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;
     uint8_t* dst = dst_buf;