[commabody] setup and run live stream encoders (#29087)

* op changes needed for body

* remove logmessaged callback

* add newlines

* add stream arg

* add stream_encoder to config

* supersets, avoid two encoderd

* fix recursive exit error

* fix loggerd crash

* add stream_encoderd cpu usage to test

* Update system/loggerd/loggerd.h

Co-authored-by: Adeeb Shihadeh <adeebshihadeh@gmail.com>

* resolve comments

* stream_encoderd onroad to false

---------

Co-authored-by: Adeeb Shihadeh <adeebshihadeh@gmail.com>

selfdrive/manager/process_config.py

@@ -13,6 +13,9 @@ def driverview(started: bool, params: Params, CP: car.CarParams) -> bool:
 def notcar(started: bool, params: Params, CP: car.CarParams) -> bool:
   return CP.notCar  # type: ignore
 
+def iscar(started: bool, params: Params, CP: car.CarParams) -> bool:
+  return not CP.notCar
+
 def logging(started, params, CP: car.CarParams) -> bool:
   run = (not CP.notCar) or not params.get_bool("DisableLogging")
   return started and run
@@ -35,12 +38,13 @@ procs = [
   NativeProcess("logcatd", "system/logcatd", ["./logcatd"]),
   NativeProcess("proclogd", "system/proclogd", ["./proclogd"]),
   PythonProcess("logmessaged", "system.logmessaged", offroad=True),
-  PythonProcess("micd", "system.micd"),
+  PythonProcess("micd", "system.micd", callback=iscar),
   PythonProcess("timezoned", "system.timezoned", enabled=not PC, offroad=True),
 
   DaemonProcess("manage_athenad", "selfdrive.athena.manage_athenad", "AthenadPid"),
   NativeProcess("dmonitoringmodeld", "selfdrive/modeld", ["./dmonitoringmodeld"], enabled=(not PC or WEBCAM), callback=driverview),
   NativeProcess("encoderd", "system/loggerd", ["./encoderd"]),
+  NativeProcess("stream_encoderd", "system/loggerd", ["./encoderd", "--stream"], onroad=False, callback=notcar),
   NativeProcess("loggerd", "system/loggerd", ["./loggerd"], onroad=False, callback=logging),
   NativeProcess("modeld", "selfdrive/modeld", ["./modeld"]),
   NativeProcess("mapsd", "selfdrive/navd", ["./mapsd"]),

system/loggerd/encoderd.cc

@@ -121,7 +121,8 @@ void encoder_thread(EncoderdState *s, const LogCameraInfo &cam_info) {
   }
 }
 
-void encoderd_thread() {
+template <size_t N>
+void encoderd_thread(const LogCameraInfo (&cameras)[N]) {
   EncoderdState s;
 
   std::set<VisionStreamType> streams;
@@ -136,9 +137,9 @@ void encoderd_thread() {
   if (!streams.empty()) {
     std::vector<std::thread> encoder_threads;
     for (auto stream : streams) {
-      auto it = std::find_if(std::begin(cameras_logged), std::end(cameras_logged),
+      auto it = std::find_if(std::begin(cameras), std::end(cameras),
                              [stream](auto &cam) { return cam.stream_type == stream; });
-      assert(it != std::end(cameras_logged));
+      assert(it != std::end(cameras));
       ++s.max_waiting;
       encoder_threads.push_back(std::thread(encoder_thread, &s, *it));
     }
@@ -147,7 +148,7 @@ void encoderd_thread() {
   }
 }
 
-int main() {
+int main(int argc, char* argv[]) {
   if (!Hardware::PC()) {
     int ret;
     ret = util::set_realtime_priority(52);
@@ -155,6 +156,15 @@ int main() {
     ret = util::set_core_affinity({3});
     assert(ret == 0);
   }
-  encoderd_thread();
-  return 0;
+  if (argc > 1) {
+    std::string arg1(argv[1]);
+    if (arg1 == "--stream") {
+      encoderd_thread(stream_cameras_logged);
+    } else {
+      LOGE("Argument '%s' is not supported", arg1.c_str());
+    }
+  } else {
+    encoderd_thread(cameras_logged);
+  }
+    return 0;  
 }

system/loggerd/loggerd.cc

@@ -205,7 +205,8 @@ void loggerd_thread() {
   // subscribe to all socks
   for (const auto& it : services) {
     const bool encoder = strcmp(it.name+strlen(it.name)-strlen("EncodeData"), "EncodeData") == 0;
-    if (!it.should_log && !encoder) continue;
+    const bool livestream_encoder = strncmp(it.name, "livestream", strlen("livestream")) == 0;
+    if (!it.should_log && (!encoder || livestream_encoder)) continue;
     LOGD("logging %s (on port %d)", it.name, it.port);
 
     SubSocket * sock = SubSocket::create(ctx.get(), it.name);

system/loggerd/loggerd.h

@@ -12,7 +12,9 @@
 #include "system/loggerd/logger.h"
 
 constexpr int MAIN_FPS = 20;
-const int MAIN_BITRATE = 10000000;
+const int MAIN_BITRATE = 1e7;
+const int LIVESTREAM_BITRATE = 1e6;
+const int QCAM_BITRATE = 256000;
 
 #define NO_CAMERA_PATIENCE 500  // fall back to time-based rotation if all cameras are dead
 
@@ -26,11 +28,10 @@ const int SEGMENT_LENGTH = LOGGERD_TEST ? atoi(getenv("LOGGERD_SEGMENT_LENGTH"))
 
 constexpr char PRESERVE_ATTR_NAME[] = "user.preserve";
 constexpr char PRESERVE_ATTR_VALUE = '1';
-
 class EncoderInfo {
 public:
   const char *publish_name;
-  const char *filename;
+  const char *filename = NULL;
   bool record = true;
   int frame_width = 1928;
   int frame_height = 1208;
@@ -57,11 +58,13 @@ const EncoderInfo main_road_encoder_info = {
   .filename = "fcamera.hevc",
   INIT_ENCODE_FUNCTIONS(RoadEncode),
 };
+
 const EncoderInfo main_wide_road_encoder_info = {
   .publish_name = "wideRoadEncodeData",
   .filename = "ecamera.hevc",
   INIT_ENCODE_FUNCTIONS(WideRoadEncode),
 };
+
 const EncoderInfo main_driver_encoder_info = {
   .publish_name = "driverEncodeData",
   .filename = "dcamera.hevc",
@@ -69,17 +72,40 @@ const EncoderInfo main_driver_encoder_info = {
   INIT_ENCODE_FUNCTIONS(DriverEncode),
 };
 
+const EncoderInfo stream_road_encoder_info = {
+  .publish_name = "livestreamRoadEncodeData",
+  .encode_type = cereal::EncodeIndex::Type::QCAMERA_H264,
+  .record = false,
+  .bitrate = LIVESTREAM_BITRATE,
+  INIT_ENCODE_FUNCTIONS(LivestreamRoadEncode),
+};
+
+const EncoderInfo stream_wide_road_encoder_info = {
+  .publish_name = "livestreamWideRoadEncodeData",
+  .encode_type = cereal::EncodeIndex::Type::QCAMERA_H264,
+  .record = false,
+  .bitrate = LIVESTREAM_BITRATE,
+  INIT_ENCODE_FUNCTIONS(LivestreamWideRoadEncode),
+};
+
+const EncoderInfo stream_driver_encoder_info = {
+  .publish_name = "livestreamDriverEncodeData",
+  .encode_type = cereal::EncodeIndex::Type::QCAMERA_H264,
+  .record = false,
+  .bitrate = LIVESTREAM_BITRATE,
+  INIT_ENCODE_FUNCTIONS(LivestreamDriverEncode),
+};
+
 const EncoderInfo qcam_encoder_info = {
   .publish_name = "qRoadEncodeData",
   .filename = "qcamera.ts",
-  .bitrate = 256000,
+  .bitrate = QCAM_BITRATE,
   .encode_type = cereal::EncodeIndex::Type::QCAMERA_H264,
   .frame_width = 526,
   .frame_height = 330,
   INIT_ENCODE_FUNCTIONS(QRoadEncode),
 };
 
-
 const LogCameraInfo road_camera_info{
   .thread_name = "road_cam_encoder",
   .type = RoadCam,
@@ -101,4 +127,26 @@ const LogCameraInfo driver_camera_info{
   .encoder_infos = {main_driver_encoder_info}
 };
 
+const LogCameraInfo stream_road_camera_info{
+  .thread_name = "road_cam_encoder",
+  .type = RoadCam,
+  .stream_type = VISION_STREAM_ROAD,
+  .encoder_infos = {stream_road_encoder_info}
+};
+
+const LogCameraInfo stream_wide_road_camera_info{
+  .thread_name = "wide_road_cam_encoder",
+  .type = WideRoadCam,
+  .stream_type = VISION_STREAM_WIDE_ROAD,
+  .encoder_infos = {stream_wide_road_encoder_info}
+};
+
+const LogCameraInfo stream_driver_camera_info{
+  .thread_name = "driver_cam_encoder",
+  .type = DriverCam,
+  .stream_type = VISION_STREAM_DRIVER,
+  .encoder_infos = {stream_driver_encoder_info}
+};
+
 const LogCameraInfo cameras_logged[] = {road_camera_info, wide_road_camera_info, driver_camera_info};
+const LogCameraInfo stream_cameras_logged[] = {stream_road_camera_info, stream_wide_road_camera_info, stream_driver_camera_info};