feat(streamer): add ffmpeg encoder and mcap recording

src/record/mcap_record_sink.cpp

@@ -0,0 +1,160 @@
+#define MCAP_IMPLEMENTATION
+#include <mcap/writer.hpp>
+
+#include "cvmmap_streamer/record/mcap_record_sink.hpp"
+
+#include "protobuf_descriptor.hpp"
+#include "foxglove/CompressedVideo.pb.h"
+
+#include <google/protobuf/timestamp.pb.h>
+
+#include <cstddef>
+#include <cstdint>
+#include <expected>
+#include <memory>
+#include <string>
+#include <string_view>
+#include <utility>
+
+namespace cvmmap_streamer::record {
+
+namespace {
+
+[[nodiscard]]
+std::string codec_format(CodecType codec) {
+	return codec == CodecType::H265 ? "h265" : "h264";
+}
+
+[[nodiscard]]
+mcap::Compression to_mcap_compression(McapCompression compression) {
+	switch (compression) {
+	case McapCompression::Lz4:
+		return mcap::Compression::Lz4;
+	case McapCompression::None:
+		return mcap::Compression::None;
+	case McapCompression::Zstd:
+	default:
+		return mcap::Compression::Zstd;
+	}
+}
+
+[[nodiscard]]
+google::protobuf::Timestamp to_proto_timestamp(std::uint64_t timestamp_ns) {
+	google::protobuf::Timestamp timestamp{};
+	timestamp.set_seconds(static_cast<std::int64_t>(timestamp_ns / 1000000000ull));
+	timestamp.set_nanos(static_cast<std::int32_t>(timestamp_ns % 1000000000ull));
+	return timestamp;
+}
+
+}
+
+struct McapRecordSink::State {
+	mcap::McapWriter writer{};
+	std::string path{};
+	std::string frame_id{};
+	mcap::ChannelId channel_id{0};
+	std::uint32_t sequence{0};
+};
+
+McapRecordSink::~McapRecordSink() {
+	close();
+}
+
+McapRecordSink::McapRecordSink(McapRecordSink &&other) noexcept
+	: state_(other.state_) {
+	other.state_ = nullptr;
+}
+
+McapRecordSink &McapRecordSink::operator=(McapRecordSink &&other) noexcept {
+	if (this != &other) {
+		close();
+		state_       = other.state_;
+		other.state_ = nullptr;
+	}
+	return *this;
+}
+
+std::expected<McapRecordSink, std::string> McapRecordSink::create(const RuntimeConfig &config) {
+	McapRecordSink sink{};
+	auto state = std::make_unique<State>();
+	state->path = config.record.mcap.path;
+	state->frame_id = config.record.mcap.frame_id;
+
+	mcap::McapWriterOptions options("");
+	options.compression = to_mcap_compression(config.record.mcap.compression);
+	const auto open_status = state->writer.open(state->path, options);
+	if (!open_status.ok()) {
+		return std::unexpected("failed to open MCAP writer at '" + state->path + "': " + open_status.message);
+	}
+
+	const auto descriptor_set = build_file_descriptor_set(foxglove::CompressedVideo::descriptor());
+	std::string schema_bytes{};
+	if (!descriptor_set.SerializeToString(&schema_bytes)) {
+		return std::unexpected("failed to serialize foxglove.CompressedVideo descriptor set");
+	}
+
+	mcap::Schema schema("foxglove.CompressedVideo", "protobuf", schema_bytes);
+	state->writer.addSchema(schema);
+
+	mcap::Channel channel(config.record.mcap.topic, "protobuf", schema.id);
+	state->writer.addChannel(channel);
+	state->channel_id = channel.id;
+
+	sink.state_ = state.release();
+	return sink;
+}
+
+std::expected<void, std::string> McapRecordSink::write_access_unit(const encode::EncodedAccessUnit &access_unit) {
+	if (state_ == nullptr) {
+		return std::unexpected("MCAP sink is not open");
+	}
+
+	foxglove::CompressedVideo message{};
+	*message.mutable_timestamp() = to_proto_timestamp(access_unit.source_timestamp_ns);
+	message.set_frame_id(state_->frame_id);
+	message.set_format(codec_format(access_unit.codec));
+	message.set_data(
+		reinterpret_cast<const char *>(access_unit.annexb_bytes.data()),
+		static_cast<int>(access_unit.annexb_bytes.size()));
+
+	std::string serialized{};
+	if (!message.SerializeToString(&serialized)) {
+		return std::unexpected("failed to serialize foxglove.CompressedVideo");
+	}
+
+	mcap::Message record{};
+	record.channelId   = state_->channel_id;
+	record.sequence    = state_->sequence++;
+	record.logTime     = access_unit.source_timestamp_ns;
+	record.publishTime = access_unit.source_timestamp_ns;
+	record.data        = reinterpret_cast<const std::byte *>(serialized.data());
+	record.dataSize    = serialized.size();
+
+	const auto write_status = state_->writer.write(record);
+	if (!write_status.ok()) {
+		return std::unexpected("failed to write MCAP message: " + write_status.message);
+	}
+	return {};
+}
+
+bool McapRecordSink::is_open() const {
+	return state_ != nullptr;
+}
+
+std::string_view McapRecordSink::path() const {
+	if (state_ == nullptr) {
+		return {};
+	}
+	return state_->path;
+}
+
+void McapRecordSink::close() {
+	if (state_ == nullptr) {
+		return;
+	}
+	state_->writer.close();
+	delete state_;
+	state_ = nullptr;
+}
+
+}

src/record/protobuf_descriptor.cpp

@@ -0,0 +1,38 @@
+#include "protobuf_descriptor.hpp"
+
+#include <queue>
+#include <string>
+#include <unordered_set>
+
+namespace cvmmap_streamer::record {
+
+google::protobuf::FileDescriptorSet build_file_descriptor_set(const google::protobuf::Descriptor *descriptor) {
+	google::protobuf::FileDescriptorSet descriptor_set;
+	if (descriptor == nullptr) {
+		return descriptor_set;
+	}
+
+	std::queue<const google::protobuf::FileDescriptor *> pending{};
+	std::unordered_set<std::string> seen{};
+	pending.push(descriptor->file());
+	seen.insert(std::string(descriptor->file()->name()));
+
+	while (!pending.empty()) {
+		const auto *next = pending.front();
+		pending.pop();
+		next->CopyTo(descriptor_set.add_file());
+		for (int index = 0; index < next->dependency_count(); ++index) {
+			const auto *dependency = next->dependency(index);
+			if (dependency == nullptr) {
+				continue;
+			}
+			if (seen.insert(std::string(dependency->name())).second) {
+				pending.push(dependency);
+			}
+		}
+	}
+
+	return descriptor_set;
+}
+
+}

src/record/protobuf_descriptor.hpp

@@ -0,0 +1,10 @@
+#pragma once
+
+#include <google/protobuf/descriptor.h>
+#include <google/protobuf/descriptor.pb.h>
+
+namespace cvmmap_streamer::record {
+
+google::protobuf::FileDescriptorSet build_file_descriptor_set(const google::protobuf::Descriptor *descriptor);
+
+}