#pragma once #include #include #include #include #include #include #include #include extern "C" { #include #include #include } #include #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 depth = 8; int fps_num = 25; int fps_den = 1; int framebuff_cnt = 3; int mxl_latency_frames = 2; std::string mxl_index_mode = "rtp"; bool direct_v210() const { return depth == 10; } enum st20_fmt transport_fmt() const { return direct_v210() ? ST20_FMT_YUV_422_10BIT : ST20_FMT_YUV_422_8BIT; } enum st_frame_fmt output_fmt() const { return direct_v210() ? ST_FRAME_FMT_V210 : ST_FRAME_FMT_UYVY; } }; inline std::string st2110_trim(std::string s) { const auto first = s.find_first_not_of(" \t\r\n"); if (first == std::string::npos) return {}; const auto last = s.find_last_not_of(" \t\r\n"); return s.substr(first, last - first + 1); } inline bool st2110_parse_int(const std::string& text, int* out) { char* end = nullptr; const long value = std::strtol(text.c_str(), &end, 10); if (!end || *end != '\0') return false; *out = static_cast(value); return true; } inline void st2110_parse_exactframerate(const std::string& value, ST2110ReceiverConfig& cfg) { const auto slash = value.find('/'); if (slash == std::string::npos) { if (!st2110_parse_int(value, &cfg.fps_num)) { throw std::runtime_error("invalid SDP exactframerate: " + value); } cfg.fps_den = 1; return; } const std::string num = value.substr(0, slash); const std::string den = value.substr(slash + 1); if (!st2110_parse_int(num, &cfg.fps_num) || !st2110_parse_int(den, &cfg.fps_den)) { throw std::runtime_error("invalid SDP exactframerate: " + value); } } inline void st2110_apply_fmtp_param(const std::string& key, const std::string& value, ST2110ReceiverConfig& cfg, std::string& sampling) { if (key == "width") { if (!st2110_parse_int(value, &cfg.width)) { throw std::runtime_error("invalid SDP width: " + value); } } else if (key == "height") { if (!st2110_parse_int(value, &cfg.height)) { throw std::runtime_error("invalid SDP height: " + value); } } else if (key == "depth") { if (!st2110_parse_int(value, &cfg.depth)) { throw std::runtime_error("invalid SDP depth: " + value); } } else if (key == "sampling") { sampling = value; } else if (key == "exactframerate") { st2110_parse_exactframerate(value, cfg); } } inline void st2110_parse_fmtp(const std::string& line, ST2110ReceiverConfig& cfg, std::string& sampling) { const auto space = line.find(' '); if (space == std::string::npos) return; std::stringstream params(line.substr(space + 1)); std::string item; while (std::getline(params, item, ';')) { item = st2110_trim(item); if (item.empty()) continue; const auto eq = item.find('='); if (eq == std::string::npos) continue; const std::string key = st2110_trim(item.substr(0, eq)); const std::string value = st2110_trim(item.substr(eq + 1)); st2110_apply_fmtp_param(key, value, cfg, sampling); } } inline void st2110_apply_sdp(const std::string& sdp, ST2110ReceiverConfig& cfg) { std::stringstream lines(sdp); std::string line; std::string sampling = "YCbCr-4:2:2"; while (std::getline(lines, line)) { line = st2110_trim(line); if (line.rfind("m=video ", 0) == 0) { std::stringstream media(line.substr(8)); int port = 0; std::string proto; int payload = 0; if (media >> port >> proto >> payload) { if (port < 0 || port > 65535 || payload < 0 || payload > 255) { throw std::runtime_error("SDP m=video port or payload out of range"); } cfg.udp_port = static_cast(port); cfg.payload_type = static_cast(payload); } } else if (line.rfind("c=IN IP4 ", 0) == 0) { std::string addr = line.substr(9); const auto slash = addr.find('/'); if (slash != std::string::npos) addr.resize(slash); cfg.mcast_ip = st2110_trim(addr); } else if (line.rfind("a=source-filter:incl IN IP4 ", 0) == 0) { std::stringstream filter(line.substr(28)); std::string group; std::string source; if (filter >> group >> source) { cfg.source_ip = source; } } else if (line.rfind("a=fmtp:", 0) == 0) { st2110_parse_fmtp(line.substr(7), cfg, sampling); } } if (sampling != "YCbCr-4:2:2" || (cfg.depth != 8 && cfg.depth != 10)) { throw std::runtime_error("only SDP YCbCr-4:2:2 depth=8 or depth=10 is supported currently"); } } 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(); if (value < 0 || value > 65535) { throw std::runtime_error(std::string(key) + " out of uint16 range"); } return static_cast(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(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(); out.ifname = cfg.at("interface").get(); out.local_ip = cfg.at("local_ip").get(); if (cfg.contains("sdp")) { st2110_apply_sdp(cfg.at("sdp").get(), out); } if (cfg.contains("source_ip")) out.source_ip = cfg.at("source_ip").get(); if (cfg.contains("mcast_ip")) out.mcast_ip = cfg.at("mcast_ip").get(); if (cfg.contains("udp_port")) out.udp_port = st2110_checked_u16(cfg, "udp_port"); out.payload_type = st2110_checked_u8(cfg, "payload_type", out.payload_type); out.width = cfg.value("width", out.width); out.height = cfg.value("height", out.height); out.depth = cfg.value("depth", out.depth); 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); out.mxl_latency_frames = cfg.value("mxl_latency_frames", out.mxl_latency_frames); out.mxl_index_mode = cfg.value("mxl_index_mode", out.mxl_index_mode); if (out.width <= 0 || out.height <= 0) { throw std::runtime_error("width and height must be positive"); } if (out.depth != 8 && out.depth != 10) { throw std::runtime_error("only depth=8 and depth=10 are supported currently"); } if (out.source_ip.empty() || out.mcast_ip.empty() || out.udp_port == 0) { throw std::runtime_error("source_ip, mcast_ip and udp_port are required unless provided by sdp"); } 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]"); } if (out.mxl_latency_frames < 1 || out.mxl_latency_frames > 30) { throw std::runtime_error("mxl_latency_frames must be in [1, 30]"); } if (out.mxl_index_mode != "rtp" && out.mxl_index_mode != "live") { throw std::runtime_error("mxl_index_mode must be 'rtp' or 'live'"); } 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(¶ms); 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(cfg.width); ops.height = static_cast(cfg.height); ops.fps = st2110_to_st_fps(cfg.fps_num, cfg.fps_den); ops.interlaced = false; ops.transport_fmt = cfg.transport_fmt(); ops.output_fmt = cfg.output_fmt(); ops.device = ST_PLUGIN_DEVICE_AUTO; ops.framebuff_cnt = static_cast(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