FrameReader: do not depend on Qtunit tests (#21379)

* add timeout and test case

* remove useless comment

* reduce to 1s

* fix gotframe fail when Duplicate POC in a sequence

* cleanup

* remove space

* fix

.github/workflows/selfdrive_tests.yaml

@@ -216,6 +216,7 @@ jobs:
                         ./selfdrive/common/tests/test_util && \
                         ./selfdrive/loggerd/tests/test_logger &&\
                         ./selfdrive/proclogd/tests/test_proclog && \
+                        ./selfdrive/ui/replay/tests/test_replay && \
                         ./selfdrive/camerad/test/ae_gray_test"
     - name: Upload coverage to Codecov
       run: bash <(curl -s https://codecov.io/bash) -v -F unit_tests

selfdrive/ui/.gitignore

@@ -4,6 +4,7 @@ moc_*
 installer/installers/*
 
 replay/replay
+replay/tests/test_replay
 qt/text
 qt/spinner
 qt/setup/setup

selfdrive/ui/SConscript

@@ -112,3 +112,6 @@ if arch in ['x86_64', 'Darwin'] and os.path.exists(Dir("#tools/").get_abspath())
   replay_lib = qt_env.Library("qt_replay", replay_lib_src, LIBS=base_libs)
   replay_libs = [replay_lib, 'avutil', 'avcodec', 'avformat', 'swscale', 'bz2'] + qt_libs
   qt_env.Program("replay/replay", ["replay/main.cc"], LIBS=replay_libs)
+
+  if GetOption('test'):
+    qt_env.Program('replay/tests/test_replay', ['replay/tests/test_replay.cc'], LIBS=[replay_libs])

selfdrive/ui/replay/framereader.cc

@@ -3,8 +3,7 @@
 #include <unistd.h>
 
 #include <cassert>
-
+#include "selfdrive/common/timing.h"
-#include <QDebug>
 
 static int ffmpeg_lockmgr_cb(void **arg, enum AVLockOp op) {
   std::mutex *mutex = (std::mutex *)*arg;
@@ -36,18 +35,13 @@ public:
   ~AVInitializer() { avformat_network_deinit(); }
 };
 
-static AVInitializer av_initializer;
+FrameReader::FrameReader(const std::string &url, int timeout_sec) : url_(url), timeout_(timeout_sec) {
-
+  static AVInitializer av_initializer;
-FrameReader::FrameReader(const std::string &url, QObject *parent) : url_(url), QObject(parent) {
-  process_thread_ = QThread::create(&FrameReader::process, this);
-  connect(process_thread_, &QThread::finished, process_thread_, &QThread::deleteLater);
-  process_thread_->start();
 }
 
 FrameReader::~FrameReader() {
   // wait until thread is finished.
   exit_ = true;
-  process_thread_->wait();
   cv_decode_.notify_all();
   cv_frame_.notify_all();
   if (decode_thread_.joinable()) {
@@ -65,25 +59,35 @@ FrameReader::~FrameReader() {
     delete[] buffer_pool.front();
     buffer_pool.pop();
   }
-  av_frame_free(&frmRgb_);
+  if (frmRgb_) {
-  avcodec_close(pCodecCtx_);
+    av_frame_free(&frmRgb_);
-  avcodec_free_context(&pCodecCtx_);
-  avformat_close_input(&pFormatCtx_);
-  sws_freeContext(sws_ctx_);
-}
-
-void FrameReader::process() {
-  if (processFrames()) {
-    decode_thread_ = std::thread(&FrameReader::decodeThread, this);
   }
-  if (!exit_) {
+  if (pCodecCtx_) {
-    emit finished();
+    avcodec_close(pCodecCtx_);
+    avcodec_free_context(&pCodecCtx_);
+  }
+  if (pFormatCtx_) {
+    avformat_close_input(&pFormatCtx_);
+  }
+  if (sws_ctx_) {
+    sws_freeContext(sws_ctx_);
   }
 }
 
-bool FrameReader::processFrames() {
+int FrameReader::check_interrupt(void *p) {
+  FrameReader *fr = static_cast<FrameReader*>(p);
+  return fr->exit_ || (fr->timeout_ > 0 && millis_since_boot() > fr->timeout_ms_);
+}
+
+bool FrameReader::process() {
+  pFormatCtx_ = avformat_alloc_context();
+  pFormatCtx_->interrupt_callback.callback = &FrameReader::check_interrupt;
+  pFormatCtx_->interrupt_callback.opaque = (void *)this;
+  if (timeout_ > 0) {
+    timeout_ms_ = millis_since_boot() + timeout_ * 1000;
+  }
   if (avformat_open_input(&pFormatCtx_, url_.c_str(), NULL, NULL) != 0) {
-    qDebug() << "error loading " << url_.c_str();
+    printf("error loading %s\n", url_.c_str());
     return false;
   }
   avformat_find_stream_info(pFormatCtx_, NULL);
@@ -91,14 +95,14 @@ bool FrameReader::processFrames() {
 
   auto pCodecCtxOrig = pFormatCtx_->streams[0]->codec;
   auto pCodec = avcodec_find_decoder(pCodecCtxOrig->codec_id);
-  assert(pCodec);
+  if (!pCodec) return false;
 
   pCodecCtx_ = avcodec_alloc_context3(pCodec);
   int ret = avcodec_copy_context(pCodecCtx_, pCodecCtxOrig);
-  assert(ret == 0);
+  if (ret != 0) return false;
 
   ret = avcodec_open2(pCodecCtx_, pCodec, NULL);
-  assert(ret >= 0);
+  if (ret < 0) return false;
 
   width = pCodecCtxOrig->width;
   height = pCodecCtxOrig->height;
@@ -106,21 +110,25 @@ bool FrameReader::processFrames() {
   sws_ctx_ = sws_getContext(width, height, AV_PIX_FMT_YUV420P,
                             width, height, AV_PIX_FMT_BGR24,
                             SWS_BILINEAR, NULL, NULL, NULL);
-  assert(sws_ctx_);
+  if (!sws_ctx_) return false;
 
   frmRgb_ = av_frame_alloc();
-  assert(frmRgb_);
+  if (!frmRgb_) return false;
 
   frames_.reserve(60 * 20);  // 20fps, one minute
   do {
     Frame &frame = frames_.emplace_back();
-    if (av_read_frame(pFormatCtx_, &frame.pkt) < 0) {
+    int err = av_read_frame(pFormatCtx_, &frame.pkt);
+    if (err < 0) {
       frames_.pop_back();
+      valid_ = (err == AVERROR_EOF);
       break;
     }
   } while (!exit_);
 
-  valid_ = !exit_;
+  if (valid_) {
+    decode_thread_ = std::thread(&FrameReader::decodeThread, this);
+  }
   return valid_;
 }
 
@@ -128,22 +136,18 @@ uint8_t *FrameReader::get(int idx) {
   if (!valid_ || idx < 0 || idx >= frames_.size()) {
     return nullptr;
   }
-
+  std::unique_lock lk(mutex_);
-  {
+  decode_idx_ = idx;
-    std::unique_lock lk(mutex_);
+  cv_decode_.notify_one();
-    decode_idx_ = idx;
+  cv_frame_.wait(lk, [=] { return exit_ || frames_[idx].data || frames_[idx].failed; });
-    cv_decode_.notify_one();
-    cv_frame_.wait(lk, [=] { return exit_ || frames_[idx].data || frames_[idx].failed; });
-  }
-
   return frames_[idx].data;
 }
 
 void FrameReader::decodeThread() {
   int idx = 0;
   while (!exit_) {
-    const int from = std::max(idx, 0);
+    const int from = std::max(idx - 15, 0);
-    const int to = std::min(from + 20, (int)frames_.size());
+    const int to = std::min(idx + 20, (int)frames_.size());
     for (int i = 0; i < frames_.size() && !exit_; ++i) {
       Frame &frame = frames_[i];
       if (i >= from && i < to) {

selfdrive/ui/replay/framereader.h

@@ -10,8 +10,6 @@
 #include <thread>
 #include <vector>
 
-#include <QThread>
-
 // independent of QT, needs ffmpeg
 extern "C" {
 #include <libavcodec/avcodec.h>
@@ -19,27 +17,22 @@ extern "C" {
 #include <libswscale/swscale.h>
 }
 
-class FrameReader : public QObject {
+class FrameReader {
-  Q_OBJECT
-
 public:
-  FrameReader(const std::string &url, QObject *parent = nullptr);
+  FrameReader(const std::string &url, int timeout_sec = 0);
   ~FrameReader();
+  bool process();
   uint8_t *get(int idx);
-  int getRGBSize() { return width * height * 3; }
+  int getRGBSize() const { return width * height * 3; }
+  size_t getFrameCount() const { return frames_.size(); }
   bool valid() const { return valid_; }
 
   int width = 0, height = 0;
 
-signals:
-  void finished();
-
 private:
-  void process();
-  bool processFrames();
   void decodeThread();
   uint8_t *decodeFrame(AVPacket *pkt);
-
+  static int check_interrupt(void *p);
   struct Frame {
     AVPacket pkt = {};
     uint8_t *data = nullptr;
@@ -47,11 +40,11 @@ private:
   };
   std::vector<Frame> frames_;
 
-  AVFormatContext *pFormatCtx_ = NULL;
+  AVFormatContext *pFormatCtx_ = nullptr;
-  AVCodecContext *pCodecCtx_ = NULL;
+  AVCodecContext *pCodecCtx_ = nullptr;
   AVFrame *frmRgb_ = nullptr;
   std::queue<uint8_t *> buffer_pool;
-  struct SwsContext *sws_ctx_ = NULL;
+  struct SwsContext *sws_ctx_ = nullptr;
 
   std::mutex mutex_;
   std::condition_variable cv_decode_;
@@ -60,6 +53,7 @@ private:
   std::atomic<bool> exit_ = false;
   bool valid_ = false;
   std::string url_;
-  QThread *process_thread_;
   std::thread decode_thread_;
+  int timeout_ = 0;
+  double timeout_ms_ = 0;
 };

selfdrive/ui/replay/replay.cc

@@ -75,6 +75,9 @@ void Replay::addSegment(int n) {
   QObject::connect(lrs[n], &LogReader::finished, this, &Replay::mergeEvents);
 
   frs[n] = new FrameReader(qPrintable(camera_paths.at(n).toString()));
+  QThread * t = QThread::create([=]() { frs[n]->process(); });
+  QObject::connect(t, &QThread::finished, t, &QThread::deleteLater);
+  t->start();
 }
 
 void Replay::removeSegment(int n) {

selfdrive/ui/replay/tests/test_replay.cc

@@ -0,0 +1,34 @@
+#define CATCH_CONFIG_MAIN
+#include "catch2/catch.hpp"
+
+#include "selfdrive/ui/replay/framereader.h"
+
+const char *stream_url = "https://commadataci.blob.core.windows.net/openpilotci/0c94aa1e1296d7c6/2021-05-05--19-48-37/0/fcamera.hevc";
+
+TEST_CASE("FrameReader") {
+  SECTION("process&get") {
+    FrameReader fr(stream_url);
+    bool ret = fr.process();
+    REQUIRE(ret == true);
+    REQUIRE(fr.valid() == true);
+    REQUIRE(fr.getFrameCount() == 1200);
+
+    // random get 50 frames
+    srand(time(NULL));
+    for (int i = 0; i < 50; ++i) {
+      int idx = rand() % (fr.getFrameCount() - 1);
+      REQUIRE(fr.get(idx) != nullptr);
+    }
+    // sequence get 50 frames {
+    for (int i = 0; i < 50; ++i) {
+      REQUIRE(fr.get(i) != nullptr);
+    }
+  }
+  SECTION("process with timeout") {
+    FrameReader fr(stream_url, 1);
+    bool ret = fr.process();
+    REQUIRE(ret == false);
+    REQUIRE(fr.valid() == false);
+    REQUIRE(fr.getFrameCount() < 1200);
+  }
+}