#include #include #include #include "V210.hpp" #include "FlowDef.hpp" #include "NodeBase.hpp" #include "VideoReader.hpp" class VideoInNode : public dmf::NodeBase { void run() override { const std::string filename = config().value("file", std::string{}); if (filename.empty()) { log("config missing 'file'"); return; } log("file: %s", filename.c_str()); dmf::VideoReader video_reader(filename); if (!video_reader.has_video && !video_reader.has_audio) { log("no video or audio stream found"); return; } const bool has_video = config().contains("video_flow_id") && video_reader.has_video; mxlFlowWriter video_writer = nullptr; mxlFlowConfigInfo video_cfg{}; uint32_t video_stride = 0; int width = 0, height = 0, fps_num = 25, fps_den = 1; std::string video_flow_id; if (has_video) { const auto video_flow_info = config().at("video_flow_id"); video_flow_id = video_flow_info.at("id").get(); width = video_flow_info.value("width", video_reader.video_info.width); height = video_flow_info.value("height", video_reader.video_info.height); fps_num = video_flow_info.value("fps_num", video_reader.video_info.fps_num); fps_den = video_flow_info.value("fps_den", video_reader.video_info.fps_den); log("video flow=%s %dx%d @ %d/%d fps", video_flow_id.c_str(), width, height, fps_num, fps_den); bool created = false; mxlStatus vst = mxlCreateFlowWriter( instance(), dmf::make_video_flow_def(video_flow_id, node_id(), width, height, fps_num, fps_den).c_str(), "", &video_writer, &video_cfg, &created); if (vst != MXL_STATUS_OK) { log("mxlCreateFlowWriter failed (%s)", dmf::mxl_status_str(vst)); return; } video_stride = video_cfg.discrete.sliceSizes[0]; log("video stride=%u B/line grain=%u B ring=%u grains", video_stride, video_stride * static_cast(height), video_cfg.discrete.grainCount); } mxlFlowWriter audio_writer = nullptr; mxlFlowConfigInfo audio_cfg{}; int sample_rate = video_reader.audio_info.sample_rate; int channels = video_reader.audio_info.channels; int bit_depth = 32; bool has_audio = config().contains("audio_flow_id") && video_reader.has_audio; int max_audio_samples = 0; if (has_audio) { const auto audio_flow_info = config().at("audio_flow_id"); const auto audio_flow_id = audio_flow_info.at("id").get(); log("audio flow=%s %d Hz %dch %d-bit", audio_flow_id.c_str(), sample_rate, channels, bit_depth); bool created = false; mxlStatus ast = mxlCreateFlowWriter( instance(), dmf::make_audio_flow_def(audio_flow_id, node_id(), sample_rate, channels, bit_depth, fps_num, fps_den).c_str(), "", &audio_writer, &audio_cfg, &created); if (ast != MXL_STATUS_OK) { log("audio mxlCreateFlowWriter failed (%s) — continuing without audio", dmf::mxl_status_str(ast)); has_audio = false; } else { log("audio channels=%u buffer=%u samples", audio_cfg.continuous.channelCount, audio_cfg.continuous.bufferLength); size_t max_write = 0; mxlFlowWriterGetMaxWriteLengthSamples(audio_writer, &max_write); max_audio_samples = static_cast(max_write); } } std::vector audio_temp(max_audio_samples * channels * sizeof(float)); const mxlRational video_rate = {fps_num, fps_den}; const mxlRational audio_rate = {sample_rate, 1}; uint64_t audio_index = 0; if (has_audio) { audio_index = mxlGetCurrentIndex(&audio_rate); } uint64_t video_index = has_video ? mxlGetCurrentIndex(&video_rate) : 0; while (dmf::g_running.load(std::memory_order_relaxed)) { uint8_t* video_buf = nullptr; mxlGrainInfo grain{}; mxlStatus vst = MXL_ERR_UNSUPPORTED_OPERATION; if (has_video) { vst = mxlFlowWriterOpenGrain(video_writer, video_index, &grain, &video_buf); } int out_samples_written = 0; dmf::VideoReader::FrameKind frame_kind = video_reader.read_next( has_video ? video_buf : nullptr, video_stride, has_audio ? audio_temp.data() : nullptr, max_audio_samples, out_samples_written ); if (frame_kind == dmf::VideoReader::FrameKind::None) break; if (frame_kind == dmf::VideoReader::FrameKind::Video) { if (has_video && vst == MXL_STATUS_OK) { grain.flags = 0; grain.validSlices = grain.totalSlices; mxlFlowWriterCommitGrain(video_writer, &grain); } if (has_video) { const uint64_t ns = mxlGetNsUntilIndex(video_index + 1, &video_rate); if (ns > 0 && ns < 2'000'000'000ULL) mxlSleepForNs(ns); video_index++; } } else if (frame_kind == dmf::VideoReader::FrameKind::Audio) { if (has_video && vst == MXL_STATUS_OK) mxlFlowWriterCancelGrain(video_writer); // Wait for audio clock to catch up — prevents TOO_EARLY and sample loss while (dmf::g_running.load(std::memory_order_relaxed)) { const uint64_t audio_now = mxlGetCurrentIndex(&audio_rate); if (audio_index + static_cast(out_samples_written) <= audio_now) break; const uint64_t ns = mxlGetNsUntilIndex(audio_index + out_samples_written, &audio_rate); if (ns > 0 && ns < 2'000'000'000ULL) mxlSleepForNs(ns); else break; } if (!dmf::g_running.load(std::memory_order_relaxed)) break; mxlMutableWrappedMultiBufferSlice slice{}; mxlStatus ast = mxlFlowWriterOpenSamples(audio_writer, audio_index, out_samples_written, &slice); if (ast != MXL_STATUS_OK) { log("audio OpenSamples failed (%s) index=%llu — skipping", dmf::mxl_status_str(ast), audio_index); audio_index += out_samples_written; // advance even on failure — keeps alignment continue; } // copy per channel from audio_temp into slice for (int ch = 0; ch < channels; ch++) { uint8_t* dst0 = static_cast(slice.base.fragments[0].pointer) + ch * slice.stride; uint8_t* src = audio_temp.data() + ch * max_audio_samples * sizeof(float); size_t frag0_bytes = slice.base.fragments[0].size; size_t total_bytes = out_samples_written * sizeof(float); if (total_bytes <= frag0_bytes) { std::memcpy(dst0, src, total_bytes); } else { // Wrapped — copy first fragment, then second std::memcpy(dst0, src, frag0_bytes); uint8_t* dst1 = static_cast(slice.base.fragments[1].pointer) + ch * slice.stride; std::memcpy(dst1, src + frag0_bytes, total_bytes - frag0_bytes); } } mxlFlowWriterCommitSamples(audio_writer); audio_index += out_samples_written; } } log("stopped at video_index=%llu", static_cast(video_index)); if (has_video) mxlReleaseFlowWriter(instance(), video_writer); if (has_audio) mxlReleaseFlowWriter(instance(), audio_writer); } }; int main() { VideoInNode node; return node.execute(); }