Revert "Encoderd: encoders just encode, no writing (#28482)"

This reverts commit eed43696f113feeac757b147a782b8c063638a63.

old-commit-hash: 5161c23645

system/loggerd/encoder/encoder.cc

@@ -38,4 +38,45 @@ void VideoEncoder::publisher_publish(VideoEncoder *e, int segment_num, uint32_t
   auto words = new kj::Array<capnp::word>(capnp::messageToFlatArray(msg));
   auto bytes = words->asBytes();
   e->pm->send(e->service_name, bytes.begin(), bytes.size());
+  if (e->write) {
+    e->to_write.push(words);
+  } else {
+    delete words;
+  }
+}
+
+// TODO: writing should be moved to loggerd
+void VideoEncoder::write_handler(VideoEncoder *e, const char *path) {
+  VideoWriter writer(path, e->filename, e->codec != cereal::EncodeIndex::Type::FULL_H_E_V_C, e->out_width, e->out_height, e->fps, e->codec);
+
+  bool first = true;
+  kj::Array<capnp::word>* out_buf;
+  while ((out_buf = e->to_write.pop())) {
+    capnp::FlatArrayMessageReader cmsg(*out_buf);
+    cereal::Event::Reader event = cmsg.getRoot<cereal::Event>();
+
+    auto edata = (e->type == DriverCam) ? event.getDriverEncodeData() :
+      ((e->type == WideRoadCam) ? event.getWideRoadEncodeData() :
+      (e->in_width == e->out_width ? event.getRoadEncodeData() : event.getQRoadEncodeData()));
+    auto idx = edata.getIdx();
+    auto flags = idx.getFlags();
+
+    if (first) {
+      assert(flags & V4L2_BUF_FLAG_KEYFRAME);
+      auto header = edata.getHeader();
+      writer.write((uint8_t *)header.begin(), header.size(), idx.getTimestampEof()/1000, true, false);
+      first = false;
+    }
+
+    // dangerous cast from const, but should be fine
+    auto data = edata.getData();
+    if (data.size() > 0) {
+      writer.write((uint8_t *)data.begin(), data.size(), idx.getTimestampEof()/1000, false, flags & V4L2_BUF_FLAG_KEYFRAME);
+    }
+
+    // free the data
+    delete out_buf;
+  }
+
+  // VideoWriter is freed on out of scope
 }

system/loggerd/encoder/encoder.h

@@ -15,9 +15,9 @@
 class VideoEncoder {
 public:
   VideoEncoder(const char* filename, CameraType type, int in_width, int in_height, int fps,
-              int bitrate, cereal::EncodeIndex::Type codec, int out_width, int out_height)
+              int bitrate, cereal::EncodeIndex::Type codec, int out_width, int out_height, bool write)
   : filename(filename), type(type), in_width(in_width), in_height(in_height), fps(fps),
-    bitrate(bitrate), codec(codec), out_width(out_width), out_height(out_height) { }
+    bitrate(bitrate), codec(codec), out_width(out_width), out_height(out_height), write(write) { }
   virtual ~VideoEncoder();
   virtual int encode_frame(VisionBuf* buf, VisionIpcBufExtra *extra) = 0;
   virtual void encoder_open(const char* path) = 0;
@@ -26,8 +26,20 @@ public:
   void publisher_init();
   static void publisher_publish(VideoEncoder *e, int segment_num, uint32_t idx, VisionIpcBufExtra &extra, unsigned int flags, kj::ArrayPtr<capnp::byte> header, kj::ArrayPtr<capnp::byte> dat);
 
+  void writer_open(const char* path) {
+    if (this->write) write_handler_thread = std::thread(VideoEncoder::write_handler, this, path);
+  }
+
+  void writer_close() {
+    if (this->write) {
+      to_write.push(NULL);
+      write_handler_thread.join();
+    }
+    assert(to_write.empty());
+  }
 
 protected:
+  bool write;
   const char* filename;
   int in_width, in_height;
   int out_width, out_height, fps;
@@ -42,4 +54,9 @@ private:
   // publishing
   std::unique_ptr<PubMaster> pm;
   const char *service_name;
+
+  // writing support
+  static void write_handler(VideoEncoder *e, const char *path);
+  std::thread write_handler_thread;
+  SafeQueue<kj::Array<capnp::word>* > to_write;
 };

system/loggerd/encoder/ffmpeg_encoder.cc

@@ -60,6 +60,7 @@ void FfmpegEncoder::encoder_open(const char* path) {
   int err = avcodec_open2(this->codec_ctx, codec, NULL);
   assert(err >= 0);
 
+  writer_open(path);
   is_open = true;
   segment_num++;
   counter = 0;
@@ -68,6 +69,7 @@ void FfmpegEncoder::encoder_open(const char* path) {
 void FfmpegEncoder::encoder_close() {
   if (!is_open) return;
 
+  writer_close();
   avcodec_free_context(&codec_ctx);
   is_open = false;
 }

system/loggerd/encoder/ffmpeg_encoder.h

@@ -17,8 +17,8 @@ extern "C" {
 class FfmpegEncoder : public VideoEncoder {
  public:
   FfmpegEncoder(const char* filename, CameraType type, int in_width, int in_height, int fps,
-                int bitrate, cereal::EncodeIndex::Type codec, int out_width, int out_height) :
-                VideoEncoder(filename, type, in_width, in_height, fps, bitrate, cereal::EncodeIndex::Type::BIG_BOX_LOSSLESS, out_width, out_height) { encoder_init(); }
+                int bitrate, cereal::EncodeIndex::Type codec, int out_width, int out_height, bool write) :
+                VideoEncoder(filename, type, in_width, in_height, fps, bitrate, cereal::EncodeIndex::Type::BIG_BOX_LOSSLESS, out_width, out_height, write) { encoder_init(); }
   ~FfmpegEncoder();
   void encoder_init();
   int encode_frame(VisionBuf* buf, VisionIpcBufExtra *extra);

system/loggerd/encoder/v4l_encoder.cc

@@ -255,6 +255,7 @@ void V4LEncoder::encoder_init() {
 
 void V4LEncoder::encoder_open(const char* path) {
   dequeue_handler_thread = std::thread(V4LEncoder::dequeue_handler, this);
+  writer_open(path);
   this->is_open = true;
   this->counter = 0;
 }
@@ -287,6 +288,7 @@ void V4LEncoder::encoder_close() {
     // join waits for V4L2_QCOM_BUF_FLAG_EOS
     dequeue_handler_thread.join();
     assert(extras.empty());
+    writer_close();
   }
   this->is_open = false;
 }

system/loggerd/encoder/v4l_encoder.h

@@ -9,8 +9,8 @@
 class V4LEncoder : public VideoEncoder {
 public:
   V4LEncoder(const char* filename, CameraType type, int in_width, int in_height, int fps,
-             int bitrate, cereal::EncodeIndex::Type codec, int out_width, int out_height) :
-             VideoEncoder(filename, type, in_width, in_height, fps, bitrate, codec, out_width, out_height) { encoder_init(); }
+             int bitrate, cereal::EncodeIndex::Type codec, int out_width, int out_height, bool write) :
+             VideoEncoder(filename, type, in_width, in_height, fps, bitrate, codec, out_width, out_height, write) { encoder_init(); }
   ~V4LEncoder();
   void encoder_init();
   int encode_frame(VisionBuf* buf, VisionIpcBufExtra *extra);

system/loggerd/encoderd.cc

@@ -57,13 +57,13 @@ void encoder_thread(EncoderdState *s, const LogCameraInfo &cam_info) {
         encoders.push_back(new Encoder(cam_info.filename, cam_info.type, buf_info.width, buf_info.height,
                                       cam_info.fps, cam_info.bitrate,
                                       cam_info.is_h265 ? cereal::EncodeIndex::Type::FULL_H_E_V_C : cereal::EncodeIndex::Type::QCAMERA_H264,
-                                      buf_info.width, buf_info.height));
+                                      buf_info.width, buf_info.height, false));
         // qcamera encoder
         if (cam_info.has_qcamera) {
           encoders.push_back(new Encoder(qcam_info.filename, cam_info.type, buf_info.width, buf_info.height,
                                         qcam_info.fps, qcam_info.bitrate,
                                         qcam_info.is_h265 ? cereal::EncodeIndex::Type::FULL_H_E_V_C : cereal::EncodeIndex::Type::QCAMERA_H264,
-                                        qcam_info.frame_width, qcam_info.frame_height));
+                                        qcam_info.frame_width, qcam_info.frame_height, false));
         }
       } else {
         LOGE("not initting empty encoder");