Loggerd for PC (#19730)

selfdrive/loggerd/SConscript

@@ -13,6 +13,7 @@ if arch in ["aarch64", "larch64"]:
   else:
     libs += ['pthread']
 else:
+  src += ['raw_logger.cc']
   libs += ['pthread']
 
 if arch == "Darwin":

selfdrive/loggerd/encoder.h

@@ -2,9 +2,6 @@
 
 #include <cstdint>
 
-#include <string>
-#include <mutex>
-
 #include "visionipc.h"
 
 class VideoEncoder {

selfdrive/loggerd/loggerd.cc

@@ -37,13 +37,13 @@
 #include "visionipc.h"
 #include "visionipc_client.h"
 
-#if !(defined(QCOM) || defined(QCOM2))
-#define DISABLE_ENCODER // TODO: loggerd for PC
-#endif
-
 #include "encoder.h"
-#ifndef DISABLE_ENCODER
+#if defined(QCOM) || defined(QCOM2)
 #include "omx_encoder.h"
+#define Encoder OmxEncoder
+#else
+#include "raw_logger.h"
+#define Encoder RawLogger
 #endif
 
 constexpr int MAIN_BITRATE = 5000000;
@@ -107,8 +107,11 @@ LogCameraInfo cameras_logged[LOG_CAMERA_ID_MAX] = {
 namespace {
 
 constexpr int SEGMENT_LENGTH = 60;
-const char* LOG_ROOT = "/data/media/0/realdata";
+#if defined(QCOM) || defined(QCOM2)
-
+std::string LOG_ROOT = "/data/media/0/realdata";
+#else
+std::string LOG_ROOT = util::getenv_default("HOME", "/.comma/media/0/realdata", "/data/media/0/realdata");
+#endif
 
 double randrange(double a, double b) __attribute__((unused));
 double randrange(double a, double b) {
@@ -189,7 +192,6 @@ struct LoggerdState {
 };
 LoggerdState s;
 
-#ifndef DISABLE_ENCODER
 void encoder_thread(int cam_idx) {
   assert(cam_idx < LOG_CAMERA_ID_MAX-1);
 
@@ -200,7 +202,7 @@ void encoder_thread(int cam_idx) {
 
   int cnt = 0;
   LoggerHandle *lh = NULL;
-  std::vector<VideoEncoder *> encoders;
+  std::vector<Encoder *> encoders;
   VisionIpcClient vipc_client = VisionIpcClient("camerad", cam_info.stream_type, false);
 
   while (!do_exit) {
@@ -209,20 +211,19 @@ void encoder_thread(int cam_idx) {
       continue;
     }
 
-    VisionBuf buf_info = vipc_client.buffers[0];
-
     // init encoders
     if (encoders.empty()) {
+      VisionBuf buf_info = vipc_client.buffers[0];
       LOGD("encoder init %dx%d", buf_info.width, buf_info.height);
 
       // main encoder
-      encoders.push_back(new OmxEncoder(cam_info.filename, buf_info.width, buf_info.height,
+      encoders.push_back(new Encoder(cam_info.filename, buf_info.width, buf_info.height,
                                      cam_info.fps, cam_info.bitrate, cam_info.is_h265, cam_info.downscale));
 
       // qcamera encoder
       if (cam_info.has_qcamera) {
         LogCameraInfo &qcam_info = cameras_logged[LOG_CAMERA_ID_QCAMERA];
-        encoders.push_back(new OmxEncoder(qcam_info.filename,
+        encoders.push_back(new Encoder(qcam_info.filename,
                                        qcam_info.frame_width, qcam_info.frame_height,
                                        qcam_info.fps, qcam_info.bitrate, qcam_info.is_h265, qcam_info.downscale));
       }
@@ -324,7 +325,6 @@ void encoder_thread(int cam_idx) {
     delete e;
   }
 }
-#endif
 
 }
 
@@ -413,7 +413,7 @@ static int clear_locks_fn(const char* fpath, const struct stat *sb, int tyupefla
 }
 
 static void clear_locks() {
-  ftw(LOG_ROOT, clear_locks_fn, 16);
+  ftw(LOG_ROOT.c_str(), clear_locks_fn, 16);
 }
 
 static void bootlog() {
@@ -425,7 +425,7 @@ static void bootlog() {
     logger_init(&s.logger, "bootlog", bytes.begin(), bytes.size(), false);
   }
 
-  err = logger_next(&s.logger, LOG_ROOT, s.segment_path, sizeof(s.segment_path), &s.rotate_segment);
+  err = logger_next(&s.logger, LOG_ROOT.c_str(), s.segment_path, sizeof(s.segment_path), &s.rotate_segment);
   assert(err == 0);
   LOGW("bootlog to %s", s.segment_path);
 
@@ -472,31 +472,31 @@ int main(int argc, char** argv) {
 
   clear_locks();
 
+  // setup messaging
   typedef struct QlogState {
     int counter, freq;
   } QlogState;
   std::map<SubSocket*, QlogState> qlog_states;
 
-  // setup messaging
-  std::vector<SubSocket*> socks;
-
   s.ctx = Context::create();
   Poller * poller = Poller::create();
+  std::vector<SubSocket*> socks;
+
+  // subscribe to all socks
   for (const auto& it : services) {
-    std::string name = it.name;
+    if (!it.should_log) continue;
 
-    if (it.should_log) {
+    SubSocket * sock = SubSocket::create(s.ctx, it.name);
-      SubSocket * sock = SubSocket::create(s.ctx, name);
     assert(sock != NULL);
     poller->registerSocket(sock);
     socks.push_back(sock);
 
     for (int cid=0; cid<=MAX_CAM_IDX; cid++) {
-        if (name == cameras_logged[cid].frame_packet_name) { s.rotate_state[cid].fpkt_sock = sock; }
+      if (std::string(it.name) == cameras_logged[cid].frame_packet_name) {
+        s.rotate_state[cid].fpkt_sock = sock;
       }
-      qlog_states[sock] = {.counter = (it.decimation == -1) ? -1 : 0,
-                           .freq = it.decimation};
     }
+    qlog_states[sock] = {.counter = 0, .freq = it.decimation};
   }
 
   // init logger
@@ -511,10 +511,10 @@ int main(int argc, char** argv) {
 
   // TODO: create these threads dynamically on frame packet presence
   std::vector<std::thread> encoder_threads;
-#ifndef DISABLE_ENCODER
   encoder_threads.push_back(std::thread(encoder_thread, LOG_CAMERA_ID_FCAMERA));
   s.rotate_state[LOG_CAMERA_ID_FCAMERA].enabled = true;
 
+#if defined(QCOM) || defined(QCOM2)
   if (record_front) {
     encoder_threads.push_back(std::thread(encoder_thread, LOG_CAMERA_ID_DCAMERA));
     s.rotate_state[LOG_CAMERA_ID_DCAMERA].enabled = true;
@@ -549,8 +549,8 @@ int main(int argc, char** argv) {
         last_msg = msg;
 
         QlogState& qs = qlog_states[sock];
-        logger_log(&s.logger, (uint8_t*)msg->getData(), msg->getSize(), qs.counter == 0);
+        logger_log(&s.logger, (uint8_t*)msg->getData(), msg->getSize(), qs.counter == 0 && qs.freq != -1);
-        if (qs.counter != -1) {
+        if (qs.freq != -1) {
           qs.counter = (qs.counter + 1) % qs.freq;
         }
 
@@ -561,7 +561,10 @@ int main(int argc, char** argv) {
       if (last_msg) {
         int fpkt_id = -1;
         for (int cid = 0; cid <=MAX_CAM_IDX; cid++) {
-          if (sock == s.rotate_state[cid].fpkt_sock) {fpkt_id=cid; break;}
+          if (sock == s.rotate_state[cid].fpkt_sock) {
+            fpkt_id=cid;
+            break;
+          }
         }
         if (fpkt_id >= 0) {
           // track camera frames to sync to encoder
@@ -617,7 +620,7 @@ int main(int argc, char** argv) {
       pthread_mutex_lock(&s.rotate_lock);
       last_rotate_tms = millis_since_boot();
 
-      int err = logger_next(&s.logger, LOG_ROOT, s.segment_path, sizeof(s.segment_path), &s.rotate_segment);
+      int err = logger_next(&s.logger, LOG_ROOT.c_str(), s.segment_path, sizeof(s.segment_path), &s.rotate_segment);
       assert(err == 0);
       if (s.logger.part == 0) {
         LOGW("logging to %s", s.segment_path);

selfdrive/loggerd/raw_logger.cc

@@ -20,11 +20,10 @@ extern "C" {
 
 #include "raw_logger.h"
 
-RawLogger::RawLogger(const std::string &afilename, int awidth, int aheight, int afps)
+RawLogger::RawLogger(const char* filename, int width, int height, int fps,
-  : filename(afilename),
+                     int bitrate, bool h265, bool downscale)
-    width(awidth),
+  : filename(filename),
-    height(aheight),
+    fps(fps) {
-    fps(afps) {
 
   int err = 0;
 
@@ -65,15 +64,16 @@ RawLogger::~RawLogger() {
   av_free(codec_ctx);
 }
 
-void RawLogger::Open(const std::string &path) {
+void RawLogger::encoder_open(const char* path, int segment) {
   int err = 0;
 
   std::lock_guard<std::recursive_mutex> guard(lock);
 
-  vid_path = util::string_format("%s/%s.mkv", path.c_str(), filename.c_str());
+  this->segment = segment;
+  vid_path = util::string_format("%s/%s.mkv", path, filename);
 
   // create camera lock file
-  lock_path = util::string_format("%s/%s.lock", path.c_str(), filename.c_str());
+  lock_path = util::string_format("%s/%s.lock", path, filename);
 
   LOG("open %s\n", lock_path.c_str());
 
@@ -105,7 +105,7 @@ void RawLogger::Open(const std::string &path) {
   counter = 0;
 }
 
-void RawLogger::Close() {
+void RawLogger::encoder_close() {
   int err = 0;
 
   std::lock_guard<std::recursive_mutex> guard(lock);
@@ -127,7 +127,9 @@ void RawLogger::Close() {
   is_open = false;
 }
 
-int RawLogger::ProcessFrame(uint64_t ts, const uint8_t *y_ptr, const uint8_t *u_ptr, const uint8_t *v_ptr) {
+int RawLogger::encode_frame(const uint8_t *y_ptr, const uint8_t *u_ptr, const uint8_t *v_ptr,
+                            int in_width, int in_height,
+                            int *frame_segment, VisionIpcBufExtra *extra) {
   int err = 0;
 
   AVPacket pkt;
@@ -138,7 +140,7 @@ int RawLogger::ProcessFrame(uint64_t ts, const uint8_t *y_ptr, const uint8_t *u_
   frame->data[0] = (uint8_t*)y_ptr;
   frame->data[1] = (uint8_t*)u_ptr;
   frame->data[2] = (uint8_t*)v_ptr;
-  frame->pts = ts;
+  frame->pts = extra->timestamp_eof;
 
   int ret = counter;
 
@@ -148,7 +150,6 @@ int RawLogger::ProcessFrame(uint64_t ts, const uint8_t *y_ptr, const uint8_t *u_
     LOGE("encoding error\n");
     ret = -1;
   } else if (got_output) {
-
     av_packet_rescale_ts(&pkt, codec_ctx->time_base, stream->time_base);
     pkt.stream_index = 0;
 

selfdrive/loggerd/raw_logger.h

@@ -15,21 +15,29 @@ extern "C" {
 #include <libavformat/avformat.h>
 }
 
-#include "frame_logger.h"
+#include "encoder.h"
 
-class RawLogger : public FrameLogger {
+class RawLogger : public VideoEncoder {
 public:
-  RawLogger(const std::string &filename, int awidth, int aheight, int afps);
+  RawLogger(const char* filename, int width, int height, int fps,
+            int bitrate, bool h265, bool downscale);
   ~RawLogger();
-
+  int encode_frame(const uint8_t *y_ptr, const uint8_t *u_ptr, const uint8_t *v_ptr,
-  int ProcessFrame(uint64_t ts, const uint8_t *y_ptr, const uint8_t *u_ptr, const uint8_t *v_ptr);
+                   int in_width, int in_height,
-  void Open(const std::string &path);
+                   int *frame_segment, VisionIpcBufExtra *extra);
-  void Close();
+  void encoder_open(const char* path, int segment);
+  void encoder_close();
 
 private:
-  std::string filename;
+  const char* filename;
-  int width, height, fps;
+  int fps;
   int counter = 0;
+  int segment = -1;
+  bool is_open = false;
+
+  std::string vid_path, lock_path;
+
+  std::recursive_mutex lock;
 
   AVCodec *codec = NULL;
   AVCodecContext *codec_ctx = NULL;

selfdrive/loggerd/tests/test_loggerd.py

@@ -15,6 +15,7 @@ from common.basedir import BASEDIR
 from common.timeout import Timeout
 from common.params import Params
 import selfdrive.manager as manager
+from selfdrive.hardware import PC
 from selfdrive.loggerd.config import ROOT
 from selfdrive.version import version as VERSION
 from tools.lib.logreader import LogReader
@@ -26,6 +27,12 @@ CEREAL_SERVICES = [f for f in log.Event.schema.union_fields if f in service_list
 
 class TestLoggerd(unittest.TestCase):
 
+  # TODO: all tests should work on PC
+  @classmethod
+  def setUpClass(cls):
+    if PC:
+      raise unittest.SkipTest
+
   def _get_latest_log_dir(self):
     log_dirs = sorted(Path(ROOT).iterdir(), key=lambda f: f.stat().st_mtime)
     return log_dirs[-1]