encoder: add ability to request keyframes from livestream encoders (#38172)

* add param that requests keyframe

* set keyframe request to false on cleanup

* clean

* timing and errors

* remove other changes

* clean diff

* clean diff

* clean up live adaptation params and functions

* better function names

* fix syntax

* remove unused var

* clean diff

common/params_keys.h

@@ -81,6 +81,7 @@ inline static std::unordered_map<std::string, ParamKeyAttributes> keys = {
     {"LiveParameters", {PERSISTENT, JSON}},
     {"LiveParametersV2", {PERSISTENT, BYTES}},
     {"LivestreamEncoderBitrate", {CLEAR_ON_MANAGER_START | DONT_LOG, INT}},
+    {"LivestreamRequestKeyframe", {CLEAR_ON_MANAGER_START | DONT_LOG, BOOL}},
     {"LiveTorqueParameters", {PERSISTENT | DONT_LOG, BYTES}},
     {"LocationFilterInitialState", {PERSISTENT, BYTES}},
     {"LateralManeuverMode", {CLEAR_ON_MANAGER_START | CLEAR_ON_OFFROAD_TRANSITION, BOOL}},

system/loggerd/encoder/encoder.h

@@ -27,6 +27,7 @@ public:
   virtual void encoder_open() = 0;
   virtual void encoder_close() = 0;
   virtual void set_bitrate(int bitrate) = 0;
+  virtual void request_keyframe() = 0;
 
   void publisher_publish(int segment_num, uint32_t idx, VisionIpcBufExtra &extra, unsigned int flags, kj::ArrayPtr<capnp::byte> header, kj::ArrayPtr<capnp::byte> dat);
 

system/loggerd/encoder/ffmpeg_encoder.cc

@@ -76,6 +76,10 @@ void FfmpegEncoder::set_bitrate(int bitrate) {
   LOGE("adaptive bitrate is not supported for ffmpeg encoder %s", encoder_info.publish_name);
 }
 
+void FfmpegEncoder::request_keyframe() {
+  LOGE("keyframe request is not supported for ffmpeg encoder %s", encoder_info.publish_name);
+}
+
 int FfmpegEncoder::encode_frame(VisionBuf* buf, VisionIpcBufExtra *extra) {
   assert(buf->width == this->in_width);
   assert(buf->height == this->in_height);

system/loggerd/encoder/ffmpeg_encoder.h

@@ -22,6 +22,7 @@ public:
   void encoder_open();
   void encoder_close();
   void set_bitrate(int bitrate);
+  void request_keyframe();
 
 private:
   int segment_num = -1;

system/loggerd/encoder/v4l_encoder.cc

@@ -156,7 +156,6 @@ V4LEncoder::V4LEncoder(const EncoderInfo &encoder_info, int in_width, int in_hei
 
   EncoderSettings encoder_settings = encoder_info.get_settings(in_width);
   current_bitrate = encoder_settings.bitrate;
-  adaptive_bitrate = encoder_info.adaptive_bitrate;
   bool is_h265 = encoder_settings.encode_type == cereal::EncodeIndex::Type::FULL_H_E_V_C;
 
   struct v4l2_format fmt_out = {
@@ -307,7 +306,7 @@ void V4LEncoder::encoder_close() {
 }
 
 void V4LEncoder::set_bitrate(int bitrate) {
-  if (!adaptive_bitrate || bitrate == current_bitrate) return;
+  if (bitrate == current_bitrate) return;
   if (bitrate <= 0) {
     LOGE("invalid livestream encoder bitrate %d", bitrate);
     return;
@@ -325,6 +324,17 @@ void V4LEncoder::set_bitrate(int bitrate) {
   current_bitrate = bitrate;
 }
 
+void V4LEncoder::request_keyframe() {
+  struct v4l2_control ctrl = {
+    .id = V4L2_CID_MPEG_VIDC_VIDEO_REQUEST_IFRAME,
+    .value = 1,
+  };
+
+  if (util::safe_ioctl(fd, VIDIOC_S_CTRL, &ctrl) == -1) {
+    LOGE("failed to request keyframe for %s", encoder_info.publish_name);
+  }
+}
+
 V4LEncoder::~V4LEncoder() {
   encoder_close();
   v4l2_buf_type buf_type = V4L2_BUF_TYPE_VIDEO_OUTPUT_MPLANE;

system/loggerd/encoder/v4l_encoder.h

@@ -14,6 +14,7 @@ public:
   void encoder_open();
   void encoder_close();
   void set_bitrate(int bitrate);
+  void request_keyframe();
 
 private:
   int fd;
@@ -22,7 +23,6 @@ private:
   int segment_num = -1;
   int counter = 0;
   int current_bitrate = -1;
-  bool adaptive_bitrate;
 
   SafeQueue<VisionIpcBufExtra> extras;
 

system/loggerd/encoderd.cc

@@ -44,14 +44,18 @@ bool sync_encoders(EncoderdState *s, VisionStreamType cam_type, uint32_t frame_i
   }
 }
 
-void apply_bitrate(std::vector<std::unique_ptr<Encoder>> &encoders) {
+void encoder_set_bitrate(std::unique_ptr<Encoder> &e) {
   static Params params;
   std::string val = params.get("LivestreamEncoderBitrate");
   if (val.empty()) return;
   int bitrate = std::stoi(val);
-  for (auto &e : encoders) {
-    e->set_bitrate(bitrate);
-  }
+  e->set_bitrate(bitrate);
+}
+
+void encoder_request_keyframe(std::unique_ptr<Encoder> &e) {
+  static Params params;
+  if (!params.getBool("LivestreamRequestKeyframe")) return;
+  e->request_keyframe();
 }
 
 void encoder_thread(EncoderdState *s, const LogCameraInfo &cam_info) {
@@ -59,9 +63,6 @@ void encoder_thread(EncoderdState *s, const LogCameraInfo &cam_info) {
 
   std::vector<std::unique_ptr<Encoder>> encoders;
 
-  bool has_adaptive = std::any_of(cam_info.encoder_infos.begin(), cam_info.encoder_infos.end(),
-                                  [](const auto &ei) { return ei.adaptive_bitrate; });
-
   VisionIpcClient vipc_client = VisionIpcClient("camerad", cam_info.stream_type, false);
 
   std::unique_ptr<JpegEncoder> jpeg_encoder;
@@ -121,10 +122,13 @@ void encoder_thread(EncoderdState *s, const LogCameraInfo &cam_info) {
         ++cur_seg;
       }
 
-      if (has_adaptive) apply_bitrate(encoders);
-
       // encode a frame
       for (int i = 0; i < encoders.size(); ++i) {
+        if (cam_info.encoder_infos[i].is_live) {
+          encoder_set_bitrate(encoders[i]);
+          encoder_request_keyframe(encoders[i]);
+        }
+
         int out_id = encoders[i]->encode_frame(buf, &extra);
 
         if (out_id == -1) {

system/loggerd/loggerd.h

@@ -59,7 +59,7 @@ public:
   const char *filename = NULL;
   bool record = true;
   bool include_audio = false;
-  bool adaptive_bitrate = false;
+  bool is_live = false;
   int frame_width = -1;
   int frame_height = -1;
   int fps = MAIN_FPS;
@@ -105,7 +105,7 @@ const EncoderInfo stream_road_encoder_info = {
   .publish_name = "livestreamRoadEncodeData",
   //.thumbnail_name = "thumbnail",
   .record = false,
-  .adaptive_bitrate = true,
+  .is_live = true,
   .get_settings = [](int){return EncoderSettings::StreamEncoderSettings();},
   INIT_ENCODE_FUNCTIONS(LivestreamRoadEncode),
 };
@@ -113,7 +113,7 @@ const EncoderInfo stream_road_encoder_info = {
 const EncoderInfo stream_wide_road_encoder_info = {
   .publish_name = "livestreamWideRoadEncodeData",
   .record = false,
-  .adaptive_bitrate = true,
+  .is_live = true,
   .get_settings = [](int){return EncoderSettings::StreamEncoderSettings();},
   INIT_ENCODE_FUNCTIONS(LivestreamWideRoadEncode),
 };
@@ -121,7 +121,7 @@ const EncoderInfo stream_wide_road_encoder_info = {
 const EncoderInfo stream_driver_encoder_info = {
   .publish_name = "livestreamDriverEncodeData",
   .record = false,
-  .adaptive_bitrate = true,
+  .is_live = true,
   .get_settings = [](int){return EncoderSettings::StreamEncoderSettings();},
   INIT_ENCODE_FUNCTIONS(LivestreamDriverEncode),
 };

system/webrtc/device/video.py

@@ -8,6 +8,11 @@ from aiortc import MediaStreamError
 
 from cereal import messaging
 from openpilot.common.realtime import DT_MDL, DT_DMON
+from openpilot.common.params import Params
+
+
+# v4l2 buffer flag marking an encoded keyframe (linux/videodev2.h)
+V4L2_BUF_FLAG_KEYFRAME = 0x8
 
 # arbitrary 16-byte UUID identifying openpilot frame-timing SEI messages
 TIMING_SEI_UUID = bytes([
@@ -32,6 +37,8 @@ class LiveStreamVideoStreamTrack(TiciVideoStreamTrack):
     self._pts = 0
     self._t0_ns = time.monotonic_ns()
     self.timing_sei_enabled = False
+    self.params = Params()
+    self._seen_keyframe = False
 
   def stop(self) -> None:
     super().stop()
@@ -63,6 +70,9 @@ class LiveStreamVideoStreamTrack(TiciVideoStreamTrack):
         raise MediaStreamError
       msg = messaging.recv_one_or_none(self._sock)
       if msg is not None:
+        if not self._seen_keyframe and (getattr(msg, msg.which()).idx.flags & V4L2_BUF_FLAG_KEYFRAME):
+          self._seen_keyframe = True
+          self.params.put("LivestreamRequestKeyframe", False, block=False)
         break
       await asyncio.sleep(0.005)
 

system/webrtc/webrtcd.py

@@ -245,6 +245,7 @@ class StreamSession:
     self.stream = builder.stream()
     self.identifier = str(uuid.uuid4())
 
+    self.params = Params()
     self.incoming_bridge: CerealIncomingMessageProxy | None = None
     self.incoming_bridge_services = incoming_services
     self.outgoing_bridge: CerealOutgoingMessageProxy | None = None
@@ -324,6 +325,7 @@ class StreamSession:
 
   async def run(self):
     try:
+      self.params.put("LivestreamRequestKeyframe", True)
       await self.stream.wait_for_connection()
       if self.stream.has_messaging_channel():
         if self.incoming_bridge is not None:
@@ -350,6 +352,7 @@ class StreamSession:
       if self._cleanup_done:
         return
       self._cleanup_done = True
+      self.params.put("LivestreamRequestKeyframe", False)
       if self.bitrate_controller is not None:
         await self.bitrate_controller.stop()
       if self.outgoing_bridge is not None: