From 34f72c22f028d57d1d9db5a29facd018f1d4ebd8 Mon Sep 17 00:00:00 2001 From: Johanness Date: Thu, 28 May 2026 23:43:26 +0300 Subject: [PATCH] hardcoded decklink mode fix --- CMakeLists.txt | 1 + decklink_test.sh | 66 +++++++++++++++++++ extern/mxl | 2 +- .../dmf-engine/node_control_client.hpp | 1 + libs/dmf-engine/src/api_server.cpp | 23 ++++++- libs/dmf-engine/src/node_control_client.cpp | 56 ++++++++++++++++ .../include/dmf-node/control_server.hpp | 2 +- libs/dmf-node/src/control_server.cpp | 11 ++-- libs/dmf-node/src/node_runner.cpp | 25 ++++--- nodes/decklink-in/src/decklink_in_node.cpp | 51 ++++++++++---- nodes/decklink-in/src/decklink_in_node.hpp | 7 +- nodes/decklink-out/src/decklink_out_node.cpp | 12 +++- 12 files changed, 222 insertions(+), 35 deletions(-) create mode 100755 decklink_test.sh diff --git a/CMakeLists.txt b/CMakeLists.txt index c25c901..12da608 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -28,6 +28,7 @@ find_package(nlohmann_json CONFIG REQUIRED) find_package(Libwebsockets CONFIG REQUIRED) find_package(Catch2 CONFIG QUIET) +set(BUILD_TESTS OFF CACHE BOOL "" FORCE) add_subdirectory(extern/mxl) add_subdirectory(libs/dmf-node) add_subdirectory(libs/dmf-engine) diff --git a/decklink_test.sh b/decklink_test.sh new file mode 100755 index 0000000..621a3e9 --- /dev/null +++ b/decklink_test.sh @@ -0,0 +1,66 @@ +#!/bin/bash +set -e + +ENGINE_PORT=${ENGINE_PORT:-9000} +BASE_URL="http://127.0.0.1:${ENGINE_PORT}" +DEVICE_INDEX=${DEVICE_INDEX:-0} +MODE=${MODE:-1080i50} + +log() { echo "=== $1 ==="; } + +log "Adding decklink-in node (device=${DEVICE_INDEX}, mode=${MODE})" +curl -s -X POST "${BASE_URL}/api/graph/nodes" \ + -H "Content-Type: application/json" \ + -d "{\"type\":\"decklink-in\",\"id\":\"sdi_in\",\"config\":{\"device_index\":${DEVICE_INDEX},\"mode\":\"${MODE}\"}}" | python3 -m json.tool 2>/dev/null || echo "" + +log "Adding passthrough node" +curl -s -X POST "${BASE_URL}/api/graph/nodes" \ + -H "Content-Type: application/json" \ + -d '{"type":"passthrough","id":"pass1"}' | python3 -m json.tool 2>/dev/null || echo "" + +log "Adding decklink-out node (device=${DEVICE_INDEX}, mode=${MODE})" +curl -s -X POST "${BASE_URL}/api/graph/nodes" \ + -H "Content-Type: application/json" \ + -d "{\"type\":\"decklink-out\",\"id\":\"sdi_out\",\"config\":{\"device_index\":${DEVICE_INDEX},\"mode\":\"${MODE}\"}}" | python3 -m json.tool 2>/dev/null || echo "" + +log "Connecting sdi_in → pass1" +curl -s -X POST "${BASE_URL}/api/graph/edges" \ + -H "Content-Type: application/json" \ + -d '{"from_node":"sdi_in","from_port":"video_out","to_node":"pass1","to_port":"video_in"}' | python3 -m json.tool 2>/dev/null || echo "" + +log "Connecting pass1 → sdi_out" +curl -s -X POST "${BASE_URL}/api/graph/edges" \ + -H "Content-Type: application/json" \ + -d '{"from_node":"pass1","from_port":"video_out","to_node":"sdi_out","to_port":"video_in"}' | python3 -m json.tool 2>/dev/null || echo "" + +log "Starting graph" +curl -s -X POST "${BASE_URL}/api/graph/start" | python3 -m json.tool 2>/dev/null || echo "" + +sleep 3 + +log "Status: sdi_in" +curl -s -X POST "${BASE_URL}/api/graph/nodes/sdi_in/command" \ + -H "Content-Type: application/json" \ + -d '{"cmd":"status"}' | python3 -m json.tool 2>/dev/null || echo "" + +log "Status: pass1" +curl -s -X POST "${BASE_URL}/api/graph/nodes/pass1/command" \ + -H "Content-Type: application/json" \ + -d '{"cmd":"status"}' | python3 -m json.tool 2>/dev/null || echo "" + +log "Status: sdi_out" +curl -s -X POST "${BASE_URL}/api/graph/nodes/sdi_out/command" \ + -H "Content-Type: application/json" \ + -d '{"cmd":"status"}' | python3 -m json.tool 2>/dev/null || echo "" + +log "Graph state" +curl -s "${BASE_URL}/api/graph" | python3 -m json.tool 2>/dev/null || echo "" + +echo "" +echo "=== SDI pipeline running. Press Enter to stop. ===" +read + +log "Stopping graph" +curl -s -X POST "${BASE_URL}/api/graph/stop" | python3 -m json.tool 2>/dev/null || echo "" + +log "Done" diff --git a/extern/mxl b/extern/mxl index 6ce7054..a0c4b6b 120000 --- a/extern/mxl +++ b/extern/mxl @@ -1 +1 @@ -../../mxl \ No newline at end of file +../mxl \ No newline at end of file diff --git a/libs/dmf-engine/include/dmf-engine/node_control_client.hpp b/libs/dmf-engine/include/dmf-engine/node_control_client.hpp index 2b5d19f..8a19e2f 100644 --- a/libs/dmf-engine/include/dmf-engine/node_control_client.hpp +++ b/libs/dmf-engine/include/dmf-engine/node_control_client.hpp @@ -9,6 +9,7 @@ namespace dmf_engine { class NodeControlClient { public: bool send_command(uint16_t port, const std::string& json_cmd); + std::string send_command_with_response(uint16_t port, const std::string& json_cmd); void register_node(const std::string& node_id, uint16_t port); void unregister_node(const std::string& node_id); diff --git a/libs/dmf-engine/src/api_server.cpp b/libs/dmf-engine/src/api_server.cpp index e338668..7d44fcb 100644 --- a/libs/dmf-engine/src/api_server.cpp +++ b/libs/dmf-engine/src/api_server.cpp @@ -254,8 +254,29 @@ static void handle_request(const std::string& method, const std::string& path, return; } + int fps_num = 50, fps_den = 1; + int width = 1920, height = 1080; + + nlohmann::json status_cmd; + status_cmd["cmd"] = "status"; + auto status_resp = cc.send_command_with_response(port_num, status_cmd.dump()); + if (!status_resp.empty()) { + try { + auto sr = nlohmann::json::parse(status_resp); + if (sr.contains("data")) { + auto& d = sr["data"]; + if (d.contains("grain_rate")) { + fps_num = d["grain_rate"].value("numerator", fps_num); + fps_den = d["grain_rate"].value("denominator", fps_den); + } + width = d.value("width", width); + height = d.value("height", height); + } + } catch (...) {} + } + auto flow_id = fm.create_flow_id(); - auto flow_def = fm.create_v210_flow_def(flow_id, 1920, 1080, 50, 1); + auto flow_def = fm.create_v210_flow_def(flow_id, width, height, fps_num, fps_den); nlohmann::json cmd; cmd["cmd"] = "add_writer"; diff --git a/libs/dmf-engine/src/node_control_client.cpp b/libs/dmf-engine/src/node_control_client.cpp index 7ae62d4..299411b 100644 --- a/libs/dmf-engine/src/node_control_client.cpp +++ b/libs/dmf-engine/src/node_control_client.cpp @@ -85,4 +85,60 @@ bool NodeControlClient::send_command(uint16_t port, const std::string& json_cmd) return ok; } +std::string NodeControlClient::send_command_with_response(uint16_t port, const std::string& json_cmd) { + int fd = socket(AF_INET, SOCK_STREAM, 0); + if (fd < 0) { + spdlog::error("NodeControlClient: socket() failed: {}", strerror(errno)); + return ""; + } + + struct timeval tv; + tv.tv_sec = 2; + tv.tv_usec = 0; + setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); + setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)); + + struct sockaddr_in addr; + std::memset(&addr, 0, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_port = htons(port); + inet_pton(AF_INET, "127.0.0.1", &addr.sin_addr); + + if (connect(fd, reinterpret_cast(&addr), sizeof(addr)) < 0) { + spdlog::error("NodeControlClient: connect to port {} failed: {}", port, strerror(errno)); + close(fd); + return ""; + } + + std::ostringstream req; + req << "POST /cmd HTTP/1.1\r\n" + << "Host: 127.0.0.1:" << port << "\r\n" + << "Content-Type: application/json\r\n" + << "Content-Length: " << json_cmd.size() << "\r\n" + << "Connection: close\r\n" + << "\r\n" + << json_cmd; + + auto request = req.str(); + auto sent = write(fd, request.data(), request.size()); + if (sent != static_cast(request.size())) { + spdlog::error("NodeControlClient: write failed on port {}", port); + close(fd); + return ""; + } + + std::string full_resp; + char resp_buf[4096]; + while (true) { + auto n = read(fd, resp_buf, sizeof(resp_buf)); + if (n <= 0) break; + full_resp.append(resp_buf, n); + } + close(fd); + + auto body_start = full_resp.find("\r\n\r\n"); + if (body_start == std::string::npos) return ""; + return full_resp.substr(body_start + 4); +} + } // namespace dmf_engine diff --git a/libs/dmf-node/include/dmf-node/control_server.hpp b/libs/dmf-node/include/dmf-node/control_server.hpp index 146423d..1cdb66d 100644 --- a/libs/dmf-node/include/dmf-node/control_server.hpp +++ b/libs/dmf-node/include/dmf-node/control_server.hpp @@ -11,7 +11,7 @@ struct lws; namespace dmf_node { -using CommandHandler = std::function; +using CommandHandler = std::function; using StatusCallback = std::function; class ControlServer { diff --git a/libs/dmf-node/src/control_server.cpp b/libs/dmf-node/src/control_server.cpp index acdd6b5..e9c06f8 100644 --- a/libs/dmf-node/src/control_server.cpp +++ b/libs/dmf-node/src/control_server.cpp @@ -29,17 +29,18 @@ struct ControlServer::Impl { struct lws_context* context = nullptr; }; -static void dispatch_command(ControlServerData* data, const nlohmann::json& msg) { +static nlohmann::json dispatch_command(ControlServerData* data, const nlohmann::json& msg) { if (!msg.contains("cmd")) { spdlog::warn("Control: message missing 'cmd' field"); - return; + return {{"error", "missing 'cmd' field"}}; } auto cmd = msg["cmd"].get(); auto it = data->commands.find(cmd); if (it != data->commands.end()) { - it->second(msg); + return it->second(msg); } else { spdlog::warn("Control: unknown command '{}'", cmd); + return {{"error", "unknown command: " + cmd}}; } } @@ -102,8 +103,8 @@ static int callback_all(struct lws* wsi, enum lws_callback_reasons reason, } try { auto msg = nlohmann::json::parse(ps->http_body); - dispatch_command(ps->data, msg); - return send_http_json(wsi, "200 OK", R"({"ok":true})"); + auto result = dispatch_command(ps->data, msg); + return send_http_json(wsi, "200 OK", result.dump()); } catch (const nlohmann::json::parse_error& e) { return send_http_json(wsi, "400 Bad Request", nlohmann::json({{"error", e.what()}}).dump()); diff --git a/libs/dmf-node/src/node_runner.cpp b/libs/dmf-node/src/node_runner.cpp index 1807b8c..51629f8 100644 --- a/libs/dmf-node/src/node_runner.cpp +++ b/libs/dmf-node/src/node_runner.cpp @@ -101,7 +101,7 @@ int NodeRunner::exec(std::unique_ptr node) { }; std::vector flow_resources; - control_server->register_command("add_writer", [&](const nlohmann::json& msg) { + control_server->register_command("add_writer", [&](const nlohmann::json& msg) -> nlohmann::json { auto flow_id = msg["flow_id"].get(); auto port_id = msg["port_id"].get(); auto flow_def = msg["flow_def"].dump(); @@ -112,14 +112,15 @@ int NodeRunner::exec(std::unique_ptr node) { auto status = mxlCreateFlowWriter(mxl_instance_, flow_def.c_str(), nullptr, &writer, &config_info, &created); if (status != MXL_STATUS_OK || !writer) { spdlog::error("Failed to create flow writer for flow {}: status={}", flow_id, static_cast(status)); - return; + return {{"ok", false}, {"error", "failed to create flow writer"}}; } spdlog::info("Created flow writer on port '{}' flow {} (created={})", port_id, flow_id, created); flow_resources.push_back({port_id, writer, nullptr}); node->on_add_writer(port_id, writer); + return {{"ok", true}}; }); - control_server->register_command("add_reader", [&](const nlohmann::json& msg) { + control_server->register_command("add_reader", [&](const nlohmann::json& msg) -> nlohmann::json { auto flow_id = msg["flow_id"].get(); auto port_id = msg["port_id"].get(); @@ -127,14 +128,15 @@ int NodeRunner::exec(std::unique_ptr node) { auto status = mxlCreateFlowReader(mxl_instance_, flow_id.c_str(), nullptr, &reader); if (status != MXL_STATUS_OK || !reader) { spdlog::error("Failed to create flow reader for flow {}: status={}", flow_id, static_cast(status)); - return; + return {{"ok", false}, {"error", "failed to create flow reader"}}; } spdlog::info("Created flow reader on port '{}' flow {}", port_id, flow_id); flow_resources.push_back({port_id, nullptr, reader}); node->on_add_reader(port_id, reader); + return {{"ok", true}}; }); - control_server->register_command("remove_writer", [&](const nlohmann::json& msg) { + control_server->register_command("remove_writer", [&](const nlohmann::json& msg) -> nlohmann::json { auto port_id = msg["port_id"].get(); node->on_remove_writer(port_id); for (auto it = flow_resources.begin(); it != flow_resources.end(); ++it) { @@ -145,9 +147,10 @@ int NodeRunner::exec(std::unique_ptr node) { } } spdlog::info("Removed writer on port '{}'", port_id); + return {{"ok", true}}; }); - control_server->register_command("remove_reader", [&](const nlohmann::json& msg) { + control_server->register_command("remove_reader", [&](const nlohmann::json& msg) -> nlohmann::json { auto port_id = msg["port_id"].get(); node->on_remove_reader(port_id); for (auto it = flow_resources.begin(); it != flow_resources.end(); ++it) { @@ -158,26 +161,30 @@ int NodeRunner::exec(std::unique_ptr node) { } } spdlog::info("Removed reader on port '{}'", port_id); + return {{"ok", true}}; }); - control_server->register_command("configure", [&](const nlohmann::json& msg) { + control_server->register_command("configure", [&](const nlohmann::json& msg) -> nlohmann::json { if (msg.contains("params")) { node->configure(msg["params"]); spdlog::info("Reconfigured node '{}'", node_id_); } + return {{"ok", true}}; }); - control_server->register_command("status", [&](const nlohmann::json& /*msg*/) { + control_server->register_command("status", [&](const nlohmann::json& /*msg*/) -> nlohmann::json { nlohmann::json resp; resp["event"] = "status"; resp["node_id"] = node_id_; resp["data"] = node->status(); control_server->send_event(resp); + return resp; }); - control_server->register_command("shutdown", [&](const nlohmann::json& /*msg*/) { + control_server->register_command("shutdown", [&](const nlohmann::json& /*msg*/) -> nlohmann::json { spdlog::info("Shutdown command received"); g_running = false; + return {{"ok", true}}; }); nlohmann::json ready_event; diff --git a/nodes/decklink-in/src/decklink_in_node.cpp b/nodes/decklink-in/src/decklink_in_node.cpp index f21eeec..7dc9e4f 100644 --- a/nodes/decklink-in/src/decklink_in_node.cpp +++ b/nodes/decklink-in/src/decklink_in_node.cpp @@ -32,6 +32,10 @@ void DeckLinkInNode::configure(const nlohmann::json& params) { spdlog::warn("DeckLink-in: unknown mode '{}', defaulting to 1080i50", mode_str); } } + + if (!open_device()) { + spdlog::error("DeckLink-in: failed to open device during configure"); + } } void DeckLinkInNode::on_add_writer(const std::string& port_id, mxlFlowWriter writer) { @@ -43,9 +47,13 @@ void DeckLinkInNode::on_add_writer(const std::string& port_id, mxlFlowWriter wri spdlog::info("DeckLink-in: writer added, grain_rate={}/{}", grain_rate_.numerator, grain_rate_.denominator); - if (!open_device()) { - spdlog::error("DeckLink-in: failed to open device"); - return; + if (input_ && !capturing_) { + if (input_->StartStreams() != S_OK) { + spdlog::error("DeckLink-in: failed to start streams"); + return; + } + capturing_ = true; + spdlog::info("DeckLink-in: capture started"); } } } @@ -107,22 +115,37 @@ bool DeckLinkInNode::open_device() { BMDTimeScale scale = 0; mode->GetFrameRate(&duration, &scale); mode->Release(); - spdlog::info("DeckLink-in: {}x{} @ {}/{} fps", frame_width_, frame_height_, duration, scale); + + is_interlaced_ = (display_mode_ == bmdModeHD1080i50 || + display_mode_ == bmdModeHD1080i5994); + + if (is_interlaced_) { + grain_rate_.numerator = static_cast(2 * scale); + grain_rate_.denominator = static_cast(duration); + } else { + grain_rate_.numerator = static_cast(scale); + grain_rate_.denominator = static_cast(duration); + } + + auto g = std::__gcd(grain_rate_.numerator, grain_rate_.denominator); + grain_rate_.numerator /= g; + grain_rate_.denominator /= g; + + spdlog::info("DeckLink-in: {}x{} @ {}/{} fps, interlaced={}", + frame_width_, frame_height_, + grain_rate_.numerator, grain_rate_.denominator, + is_interlaced_); } - if (input_->StartStreams() != S_OK) { - spdlog::error("DeckLink-in: failed to start streams"); - return false; - } - - capturing_ = true; - spdlog::info("DeckLink-in: capture started"); + spdlog::info("DeckLink-in: device opened, waiting for writer to start capture"); return true; } void DeckLinkInNode::close_device() { if (input_) { - input_->StopStreams(); + if (capturing_) { + input_->StopStreams(); + } input_->DisableVideoInput(); input_->Release(); input_ = nullptr; @@ -199,6 +222,10 @@ nlohmann::json DeckLinkInNode::status() const { {"write_index", write_index_}, {"capturing", capturing_.load()}, {"has_writer", writer_.has_value()}, + {"grain_rate", {{"numerator", grain_rate_.numerator}, {"denominator", grain_rate_.denominator}}}, + {"width", frame_width_}, + {"height", frame_height_}, + {"interlaced", is_interlaced_}, }; } diff --git a/nodes/decklink-in/src/decklink_in_node.hpp b/nodes/decklink-in/src/decklink_in_node.hpp index e55ec5b..3cc47a5 100644 --- a/nodes/decklink-in/src/decklink_in_node.hpp +++ b/nodes/decklink-in/src/decklink_in_node.hpp @@ -40,7 +40,7 @@ private: void close_device(); std::optional writer_; - mxlRational grain_rate_{50, 1}; + mxlRational grain_rate_{25, 1}; uint64_t write_index_ = 0; uint64_t grains_written_ = 0; @@ -51,11 +51,12 @@ private: IDeckLinkInput* input_ = nullptr; std::atomic capturing_{false}; + bool is_interlaced_ = false; std::mutex frame_mutex_; void* frame_data_ = nullptr; long frame_row_bytes_ = 0; - long frame_width_ = 0; - long frame_height_ = 0; + long frame_width_ = 1920; + long frame_height_ = 1080; bool frame_ready_ = false; class CaptureCallback : public IDeckLinkInputCallback { diff --git a/nodes/decklink-out/src/decklink_out_node.cpp b/nodes/decklink-out/src/decklink_out_node.cpp index f22e259..b54127a 100644 --- a/nodes/decklink-out/src/decklink_out_node.cpp +++ b/nodes/decklink-out/src/decklink_out_node.cpp @@ -201,10 +201,15 @@ void DeckLinkOutNode::process() { mxlFlowConfigInfo config{}; mxlFlowReaderGetConfigInfo(*reader_, &config); auto grain_size = grain_info.grainSize; - auto height = config.discrete.grainCount > 0 ? 1080 : 1080; - auto row_bytes = grain_size / height; - schedule_frame(payload, 1920, height, row_bytes); + long height = 1080; + if (config.discrete.sliceSizes[0] > 0) { + height = grain_size / config.discrete.sliceSizes[0]; + } + long row_bytes = (config.discrete.sliceSizes[0] > 0) ? static_cast(config.discrete.sliceSizes[0]) : (grain_size / height); + long width = (row_bytes * 3) / 8; + + schedule_frame(payload, width, height, row_bytes); read_index_ = grain_info.index + 1; grains_read_++; @@ -221,6 +226,7 @@ nlohmann::json DeckLinkOutNode::status() const { {"read_index", read_index_}, {"playing", playing_.load()}, {"has_reader", reader_.has_value()}, + {"grain_rate", {{"numerator", grain_rate_.numerator}, {"denominator", grain_rate_.denominator}}}, }; }