canbana: complete basic functions (#25965)

* add chart header

* get all signal val from logs

* loop in selected range

* clear list before append

* automatically zoom on yaxis

* cleanup

* sync charts

* fix event_begin_sec

* set the color of rubber

* add TODO

* sync slider with charts

* keep video aspect ratio

* sync plot buttons

* reduce flickers

* cleanup

* refactor detail view

* clear counters

* more

use qcamera
old-commit-hash: a6ba073231761e06ac6f070a01b434243d9d0693

tools/replay/logreader.cc

@@ -46,7 +46,9 @@ LogReader::~LogReader() {
 #endif
 }
 
-bool LogReader::load(const std::string &url, std::atomic<bool> *abort, bool local_cache, int chunk_size, int retries) {
+bool LogReader::load(const std::string &url, std::atomic<bool> *abort,
+                     const std::set<cereal::Event::Which> &allow,
+                     bool local_cache, int chunk_size, int retries) {
   raw_ = FileReader(local_cache, chunk_size, retries).read(url, abort);
   if (raw_.empty()) return false;
 
@@ -54,18 +56,26 @@ bool LogReader::load(const std::string &url, std::atomic<bool> *abort, bool loca
     raw_ = decompressBZ2(raw_, abort);
     if (raw_.empty()) return false;
   }
-  return parse(abort);
+  return parse(allow, abort);
 }
 
 bool LogReader::load(const std::byte *data, size_t size, std::atomic<bool> *abort) {
   raw_.assign((const char *)data, size);
-  return parse(abort);
+  return parse({}, abort);
 }
 
-bool LogReader::parse(std::atomic<bool> *abort) {
+bool LogReader::parse(const std::set<cereal::Event::Which> &allow, std::atomic<bool> *abort) {
   try {
     kj::ArrayPtr<const capnp::word> words((const capnp::word *)raw_.data(), raw_.size() / sizeof(capnp::word));
     while (words.size() > 0 && !(abort && *abort)) {
+      if (!allow.empty()) {
+        capnp::FlatArrayMessageReader reader(words);
+        auto which = reader.getRoot<cereal::Event>().which();
+        if (allow.find(which) == allow.end()) {
+          words = kj::arrayPtr(reader.getEnd(), words.end());
+          continue;
+        }
+      }
 
 #ifdef HAS_MEMORY_RESOURCE
       Event *evt = new (mbr_) Event(words);

tools/replay/logreader.h

@@ -5,6 +5,8 @@
 #include <memory_resource>
 #endif
 
+#include <set>
+
 #include "cereal/gen/cpp/log.capnp.h"
 #include "system/camerad/cameras/camera_common.h"
 #include "tools/replay/filereader.h"
@@ -50,12 +52,13 @@ class LogReader {
 public:
   LogReader(size_t memory_pool_block_size = DEFAULT_EVENT_MEMORY_POOL_BLOCK_SIZE);
   ~LogReader();
-  bool load(const std::string &url, std::atomic<bool> *abort = nullptr, bool local_cache = false, int chunk_size = -1, int retries = 0);
+  bool load(const std::string &url, std::atomic<bool> *abort = nullptr, const std::set<cereal::Event::Which> &allow = {},
+            bool local_cache = false, int chunk_size = -1, int retries = 0);
   bool load(const std::byte *data, size_t size, std::atomic<bool> *abort = nullptr);
   std::vector<Event*> events;
 
 private:
-  bool parse(std::atomic<bool> *abort);
+  bool parse(const std::set<cereal::Event::Which> &allow, std::atomic<bool> *abort);
   std::string raw_;
 #ifdef HAS_MEMORY_RESOURCE
   std::pmr::monotonic_buffer_resource *mbr_ = nullptr;

tools/replay/replay.cc

@@ -19,6 +19,9 @@ Replay::Replay(QString route, QStringList allow, QStringList block, SubMaster *s
     if ((allow.empty() || allow.contains(it.name)) && !block.contains(it.name)) {
       uint16_t which = event_struct.getFieldByName(it.name).getProto().getDiscriminantValue();
       sockets_[which] = it.name;
+      if (!allow.empty() || !block.empty()) {
+        allow_list.insert((cereal::Event::Which)which);
+      }
       s.push_back(it.name);
     }
   }
@@ -91,17 +94,17 @@ void Replay::updateEvents(const std::function<bool()> &lambda) {
   stream_cv_.notify_one();
 }
 
-void Replay::seekTo(int seconds, bool relative) {
+void Replay::seekTo(double seconds, bool relative) {
   seconds = relative ? seconds + currentSeconds() : seconds;
   updateEvents([&]() {
-    seconds = std::max(0, seconds);
-    int seg = seconds / 60;
+    seconds = std::max(double(0.0), seconds);
+    int seg = (int)seconds / 60;
     if (segments_.find(seg) == segments_.end()) {
       rWarning("can't seek to %d s segment %d is invalid", seconds, seg);
       return true;
     }
 
-    rInfo("seeking to %d s, segment %d", seconds, seg);
+    rInfo("seeking to %d s, segment %d", (int)seconds, seg);
     current_segment_ = seg;
     cur_mono_time_ = route_start_ts_ + seconds * 1e9;
     return isSegmentMerged(seg);
@@ -122,7 +125,9 @@ void Replay::buildTimeline() {
 
   for (int i = 0; i < segments_.size() && !exit_; ++i) {
     LogReader log;
-    if (!log.load(route_->at(i).qlog.toStdString(), &exit_, !hasFlag(REPLAY_FLAG_NO_FILE_CACHE), 0, 3)) continue;
+    if (!log.load(route_->at(i).qlog.toStdString(), &exit_,
+                  {cereal::Event::Which::CONTROLS_STATE, cereal::Event::Which::USER_FLAG},
+                  !hasFlag(REPLAY_FLAG_NO_FILE_CACHE), 0, 3)) continue;
 
     for (const Event *e : log.events) {
       if (e->which == cereal::Event::Which::CONTROLS_STATE) {
@@ -215,7 +220,7 @@ void Replay::queueSegment() {
     if ((seg && !seg->isLoaded()) || !seg) {
       if (!seg) {
         rDebug("loading segment %d...", n);
-        seg = std::make_unique<Segment>(n, route_->at(n), flags_);
+        seg = std::make_unique<Segment>(n, route_->at(n), flags_, allow_list);
         QObject::connect(seg.get(), &Segment::loadFinished, this, &Replay::segmentLoadFinished);
       }
       break;
@@ -270,6 +275,9 @@ void Replay::mergeSegments(const SegmentMap::iterator &begin, const SegmentMap::
       segments_merged_ = segments_need_merge;
       return true;
     });
+    if (stream_thread_) {
+      emit segmentsMerged();
+    }
   }
 }
 
@@ -306,6 +314,7 @@ void Replay::startStream(const Segment *cur_segment) {
     camera_server_ = std::make_unique<CameraServer>(camera_size);
   }
 
+  emit segmentsMerged();
   // start stream thread
   stream_thread_ = new QThread();
   QObject::connect(stream_thread_, &QThread::started, [=]() { stream(); });

tools/replay/replay.h

@@ -45,18 +45,19 @@ public:
   void stop();
   void pause(bool pause);
   void seekToFlag(FindFlag flag);
-  void seekTo(int seconds, bool relative);
+  void seekTo(double seconds, bool relative);
   inline bool isPaused() const { return paused_; }
   inline bool hasFlag(REPLAY_FLAGS flag) const { return flags_ & flag; }
   inline void addFlag(REPLAY_FLAGS flag) { flags_ |= flag; }
   inline void removeFlag(REPLAY_FLAGS flag) { flags_ &= ~flag; }
   inline const Route* route() const { return route_.get(); }
-  inline int currentSeconds() const { return (cur_mono_time_ - route_start_ts_) / 1e9; }
+  inline double currentSeconds() const { return double(cur_mono_time_ - route_start_ts_) / 1e9; }
   inline uint64_t routeStartTime() const { return route_start_ts_; }
   inline int toSeconds(uint64_t mono_time) const { return (mono_time - route_start_ts_) / 1e9; }
   inline int totalSeconds() const { return segments_.size() * 60; }
   inline void setSpeed(float speed) { speed_ = speed; }
   inline float getSpeed() const { return speed_; }
+  inline const std::vector<Event *> *events() const { return events_.get(); }
   inline const std::string &carFingerprint() const { return car_fingerprint_; }
   inline const std::vector<std::tuple<int, int, TimelineType>> getTimeline() {
     std::lock_guard lk(timeline_lock);
@@ -65,6 +66,7 @@ public:
 
 signals:
   void streamStarted();
+  void segmentsMerged();
 
 protected slots:
   void segmentLoadFinished(bool success);
@@ -98,7 +100,7 @@ protected:
   bool paused_ = false;
   bool events_updated_ = false;
   uint64_t route_start_ts_ = 0;
-  uint64_t cur_mono_time_ = 0;
+  std::atomic<uint64_t> cur_mono_time_ = 0;
   std::unique_ptr<std::vector<Event *>> events_;
   std::unique_ptr<std::vector<Event *>> new_events_;
   std::vector<int> segments_merged_;
@@ -114,6 +116,7 @@ protected:
   std::mutex timeline_lock;
   QFuture<void> timeline_future;
   std::vector<std::tuple<int, int, TimelineType>> timeline;
+  std::set<cereal::Event::Which> allow_list;
   std::string car_fingerprint_;
   float speed_ = 1.0;
 };

tools/replay/route.cc

@@ -99,7 +99,9 @@ void Route::addFileToSegment(int n, const QString &file) {
 
 // class Segment
 
-Segment::Segment(int n, const SegmentFile &files, uint32_t flags) : seg_num(n), flags(flags) {
+Segment::Segment(int n, const SegmentFile &files, uint32_t flags,
+                 const std::set<cereal::Event::Which> &allow)
+    : seg_num(n), flags(flags), allow(allow) {
   // [RoadCam, DriverCam, WideRoadCam, log]. fallback to qcamera/qlog
   const std::array file_list = {
       (flags & REPLAY_FLAG_QCAMERA) || files.road_cam.isEmpty() ? files.qcamera : files.road_cam,
@@ -130,7 +132,7 @@ void Segment::loadFile(int id, const std::string file) {
     success = frames[id]->load(file, flags & REPLAY_FLAG_NO_HW_DECODER, &abort_, local_cache, 20 * 1024 * 1024, 3);
   } else {
     log = std::make_unique<LogReader>();
-    success = log->load(file, &abort_, local_cache, 0, 3);
+    success = log->load(file, &abort_, allow, local_cache, 0, 3);
   }
 
   if (!success) {

tools/replay/route.h

@@ -47,7 +47,7 @@ class Segment : public QObject {
   Q_OBJECT
 
 public:
-  Segment(int n, const SegmentFile &files, uint32_t flags);
+  Segment(int n, const SegmentFile &files, uint32_t flags, const std::set<cereal::Event::Which> &allow = {});
   ~Segment();
   inline bool isLoaded() const { return !loading_ && !abort_; }
 
@@ -65,4 +65,5 @@ protected:
   std::atomic<int> loading_ = 0;
   QFutureSynchronizer<void> synchronizer_;
   uint32_t flags;
+  std::set<cereal::Event::Which> allow;
 };