Cabana: support live streaming (#26946)

* support live streaming

* update live stream's time

* cleanup stream classes

* disable video control in live streaming mode

* emit streamStarted() in LiveStream::streamThread

* disable some features in live streaming mode

* refactor charts to support live streaming mode

* disable dynamic mode checkbox in live streaming mode

* updateDispalyRange

* thread safe events

* TODO: add support for ZMQ

* atomic time stamp

* only keep settings.cached_segment_limit*60  seconds data in liveStream

* make charts work better in live mode

* cleanup ChartView

* fix toolbar

* cleanup

cleanup

* disable openpilotPrefix and useOpenGL on macos

* add comment

* exit gracefully

* support ZMQ

* use ::operator new/delete

* cleanup streams

* cleanup

* align stream buffers

* check looping back

* check if series is empty

* cleanup

* add TODO: write stream to log file to replay it

* upper_bound

* remove class member event_range

* change default settings value

* cleanup updateDisplayrange

* fix merge error
old-commit-hash: f9490739ab

tools/cabana/SConscript

@@ -20,8 +20,8 @@ cabana_env = qt_env.Clone()
 
 prev_moc_path = cabana_env['QT_MOCHPREFIX']
 cabana_env['QT_MOCHPREFIX'] = os.path.dirname(prev_moc_path) + '/cabana/moc_'
-cabana_lib = cabana_env.Library("cabana_lib", ['mainwin.cc', 'binaryview.cc', 'chartswidget.cc', 'historylog.cc', 'videowidget.cc', 'signaledit.cc', 'dbcmanager.cc',
+cabana_lib = cabana_env.Library("cabana_lib", ['mainwin.cc', 'streams/livestream.cc', 'streams/abstractstream.cc', 'streams/replaystream.cc', 'binaryview.cc', 'chartswidget.cc', 'historylog.cc', 'videowidget.cc', 'signaledit.cc', 'dbcmanager.cc',
-                            'canmessages.cc', 'commands.cc', 'messageswidget.cc', 'settings.cc', 'detailwidget.cc', 'tools/findsimilarbits.cc'], LIBS=cabana_libs, FRAMEWORKS=base_frameworks)
+                                               'commands.cc', 'messageswidget.cc', 'settings.cc', 'detailwidget.cc', 'tools/findsimilarbits.cc'], LIBS=cabana_libs, FRAMEWORKS=base_frameworks)
 cabana_env.Program('_cabana', ['cabana.cc', cabana_lib, asset_obj], LIBS=cabana_libs, FRAMEWORKS=base_frameworks)
 
 if GetOption('test'):

tools/cabana/binaryview.cc

@@ -6,7 +6,7 @@
 #include <QPainter>
 #include <QToolTip>
 
-#include "tools/cabana/canmessages.h"
+#include "tools/cabana/streams/abstractstream.h"
 
 // BinaryView
 

tools/cabana/cabana.cc

@@ -4,6 +4,8 @@
 #include "common/prefix.h"
 #include "selfdrive/ui/qt/util.h"
 #include "tools/cabana/mainwin.h"
+#include "tools/cabana/streams/livestream.h"
+#include "tools/cabana/streams/replaystream.h"
 
 int main(int argc, char *argv[]) {
   QCoreApplication::setApplicationName("Cabana");
@@ -17,33 +19,40 @@ int main(int argc, char *argv[]) {
   cmd_parser.addOption({"demo", "use a demo route instead of providing your own"});
   cmd_parser.addOption({"qcam", "load qcamera"});
   cmd_parser.addOption({"ecam", "load wide road camera"});
+  cmd_parser.addOption({"stream", "read can messages from live streaming"});
+  cmd_parser.addOption({"zmq", "the ip address on which to receive zmq messages", "zmq"});
   cmd_parser.addOption({"data_dir", "local directory with routes", "data_dir"});
   cmd_parser.process(app);
   const QStringList args = cmd_parser.positionalArguments();
-  if (args.empty() && !cmd_parser.isSet("demo")) {
+  if (args.empty() && !cmd_parser.isSet("demo") && !cmd_parser.isSet("stream")) {
     cmd_parser.showHelp();
   }
 
+  std::unique_ptr<OpenpilotPrefix> op_prefix;
+  std::unique_ptr<AbstractStream> stream;
 
-  const QString route = args.empty() ? DEMO_ROUTE : args.first();
+  if (cmd_parser.isSet("stream")) {
-  uint32_t replay_flags = REPLAY_FLAG_NONE;
+    stream.reset(new LiveStream(&app, cmd_parser.value("zmq")));
-  if (cmd_parser.isSet("ecam")) {
+  } else {
-    replay_flags |= REPLAY_FLAG_ECAM;
+    // TODO: Remove when OpenpilotPrefix supports ZMQ
-  } else if (cmd_parser.isSet("qcam")) {
-    replay_flags |= REPLAY_FLAG_QCAMERA;
-  }
-
-  // TODO: Remove when OpenpilotPrefix supports ZMQ
 #ifndef __APPLE__
-  OpenpilotPrefix op_prefix;
+    op_prefix.reset(new OpenpilotPrefix());
 #endif
-
+    const QString route = args.empty() ? DEMO_ROUTE : args.first();
-  CANMessages p(&app);
+    uint32_t replay_flags = REPLAY_FLAG_NONE;
-  int ret = 0;
+    if (cmd_parser.isSet("ecam")) {
-  if (p.loadRoute(route, cmd_parser.value("data_dir"), replay_flags)) {
+      replay_flags |= REPLAY_FLAG_ECAM;
-    MainWindow w;
+    } else if (cmd_parser.isSet("qcam")) {
-    w.show();
+      replay_flags |= REPLAY_FLAG_QCAMERA;
-    ret = app.exec();
+    }
+    auto replay_stream = new ReplayStream(&app);
+    stream.reset(replay_stream);
+    if (!replay_stream->loadRoute(route, cmd_parser.value("data_dir"), replay_flags)) {
+      return 0;
+    }
   }
-  return ret;
+
+  MainWindow w;
+  w.show();
+  return app.exec();
 }

tools/cabana/canmessages.h

@@ -1,86 +0,0 @@
-#pragma once
-
-#include <atomic>
-
-#include <QColor>
-#include <QHash>
-#include <QApplication>
-
-#include "opendbc/can/common_dbc.h"
-#include "tools/cabana/settings.h"
-#include "tools/replay/replay.h"
-
-struct CanData {
-  double ts = 0.;
-  uint8_t src = 0;
-  uint32_t address = 0;
-  uint32_t count = 0;
-  uint32_t freq = 0;
-  QByteArray dat;
-  QList<QColor> colors;
-};
-
-class CANMessages : public QObject {
-  Q_OBJECT
-
-public:
-  CANMessages(QObject *parent);
-  ~CANMessages();
-  bool loadRoute(const QString &route, const QString &data_dir, uint32_t replay_flags = REPLAY_FLAG_NONE);
-  void seekTo(double ts);
-  bool eventFilter(const Event *event);
-
-  inline QString routeName() const { return replay->route()->name(); }
-  inline QString carFingerprint() const { return replay->carFingerprint().c_str(); }
-  inline VisionStreamType visionStreamType() const { return replay->hasFlag(REPLAY_FLAG_ECAM) ? VISION_STREAM_WIDE_ROAD : VISION_STREAM_ROAD; }
-  inline double totalSeconds() const { return replay->totalSeconds(); }
-  inline double routeStartTime() const { return replay->routeStartTime() / (double)1e9; }
-  inline double currentSec() const { return replay->currentSeconds(); }
-  inline QDateTime currentDateTime() const { return replay->currentDateTime(); }
-  inline const CanData &lastMessage(const QString &id) { return can_msgs[id]; }
-
-  inline const Route* route() const { return replay->route(); }
-  inline const std::vector<Event *> *events() const { return replay->events(); }
-  inline void setSpeed(float speed) { replay->setSpeed(speed); }
-  inline bool isPaused() const { return replay->isPaused(); }
-  void pause(bool pause);
-  inline const std::vector<std::tuple<int, int, TimelineType>> getTimeline() { return replay->getTimeline(); }
-
-signals:
-  void paused();
-  void resume();
-  void seekedTo(double sec);
-  void streamStarted();
-  void eventsMerged();
-  void updated();
-  void msgsReceived(const QHash<QString, CanData> *);
-  void received(QHash<QString, CanData> *);
-
-public:
-  QMap<QString, CanData> can_msgs;
-
-protected:
-  void process(QHash<QString, CanData> *);
-  void settingChanged();
-
-  Replay *replay = nullptr;
-  std::atomic<double> counters_begin_sec = 0;
-  std::atomic<bool> processing = false;
-  QHash<QString, uint32_t> counters;
-};
-
-inline QString toHex(const QByteArray &dat) {
-  return dat.toHex(' ').toUpper();
-}
-inline char toHex(uint value) {
-  return "0123456789ABCDEF"[value & 0xF];
-}
-
-inline const QString &getColor(int i) {
-  // TODO: add more colors
-  static const QString SIGNAL_COLORS[] = {"#9FE2BF", "#40E0D0", "#6495ED", "#CCCCFF", "#FF7F50", "#FFBF00"};
-  return SIGNAL_COLORS[i % std::size(SIGNAL_COLORS)];
-}
-
-// A global pointer referring to the unique CANMessages object
-extern CANMessages *can;

tools/cabana/chartswidget.cc

@@ -69,8 +69,8 @@ ChartsWidget::ChartsWidget(QWidget *parent) : QWidget(parent) {
   column_count = settings.chart_column_count;
 
   QObject::connect(dbc(), &DBCManager::DBCFileChanged, this, &ChartsWidget::removeAll);
-  QObject::connect(can, &CANMessages::eventsMerged, this, &ChartsWidget::eventsMerged);
+  QObject::connect(can, &AbstractStream::eventsMerged, this, &ChartsWidget::updateState);
-  QObject::connect(can, &CANMessages::updated, this, &ChartsWidget::updateState);
+  QObject::connect(can, &AbstractStream::updated, this, &ChartsWidget::updateState);
   QObject::connect(show_all_values_btn, &QAction::triggered, this, &ChartsWidget::showAllData);
   QObject::connect(remove_all_btn, &QAction::triggered, this, &ChartsWidget::removeAll);
   QObject::connect(reset_zoom_btn, &QAction::triggered, this, &ChartsWidget::zoomReset);
@@ -83,27 +83,33 @@ ChartsWidget::ChartsWidget(QWidget *parent) : QWidget(parent) {
   });
 }
 
-void ChartsWidget::eventsMerged() {
-  if (auto events = can->events(); events && !events->empty()) {
-    event_range.first = (events->front()->mono_time / (double)1e9) - can->routeStartTime();
-    event_range.second = (events->back()->mono_time / (double)1e9) - can->routeStartTime();
-    updateState();
-  }
-}
-
 void ChartsWidget::updateDisplayRange() {
-  auto prev_range = display_range;
+  auto events = can->events();
-  double current_sec = can->currentSec();
+  double min_event_sec = (events->front()->mono_time / (double)1e9) - can->routeStartTime();
-  if (current_sec < display_range.first || current_sec >= (display_range.second - 5)) {
+  double max_event_sec = (events->back()->mono_time / (double)1e9) - can->routeStartTime();
-    // reached the end, or seeked to a timestamp out of range.
+  const double cur_sec = can->currentSec();
-    display_range.first = current_sec - 5;
+  if (!can->liveStreaming()) {
-  }
+    auto prev_range = display_range;
-  display_range.first = std::floor(std::max(display_range.first, event_range.first) * 10.0) / 10.0;
+    if (cur_sec < display_range.first || cur_sec >= (display_range.second - 5)) {
-  display_range.second = std::floor(std::min(display_range.first + max_chart_range, event_range.second) * 10.0) / 10.0;
+      // reached the end, or seeked to a timestamp out of range.
-  if (prev_range != display_range) {
+      display_range.first = cur_sec - 5;
-    QFutureSynchronizer<void> future_synchronizer;
+    }
-    for (auto c : charts)
+    display_range.first = std::floor(std::max(min_event_sec, display_range.first));
-      future_synchronizer.addFuture(QtConcurrent::run(c, &ChartView::setEventsRange, display_range));
+    display_range.second = std::floor(std::min(display_range.first + max_chart_range, max_event_sec));
+    if (prev_range != display_range) {
+      QFutureSynchronizer<void> future_synchronizer;
+      for (auto c : charts) {
+        future_synchronizer.addFuture(QtConcurrent::run(c, &ChartView::setEventsRange, display_range));
+      }
+    }
+  } else {
+    if (cur_sec >= (display_range.second - 5)) {
+      display_range.first = std::floor(std::max(min_event_sec, cur_sec - settings.max_chart_x_range / 2.0));
+    }
+    display_range.second = std::floor(display_range.first + settings.max_chart_x_range);
+    for (auto c : charts) {
+      c->updateSeries(nullptr, events, false);
+    }
   }
 }
 
@@ -151,7 +157,7 @@ void ChartsWidget::updateToolBar() {
   bool displaying_all = max_chart_range != min_range;
   show_all_values_btn->setText(tr("%1 minutes").arg(max_chart_range / 60));
   show_all_values_btn->setToolTip(tr("Click to display %1 data").arg(displaying_all ? tr("%1 minutes").arg(min_range / 60) : tr("ALL cached")));
-  show_all_values_btn->setVisible(!is_zoomed);
+  show_all_values_btn->setVisible(!is_zoomed && !can->liveStreaming());
   remove_all_btn->setEnabled(!charts.isEmpty());
   reset_zoom_btn->setEnabled(is_zoomed);
   range_label->setText(is_zoomed ? tr("%1 - %2").arg(zoomed_range.first, 0, 'f', 2).arg(zoomed_range.second, 0, 'f', 2) : "");
@@ -274,6 +280,7 @@ ChartView::ChartView(QWidget *parent) : QChartView(nullptr, parent) {
   chart->addAxis(axis_y, Qt::AlignLeft);
   chart->legend()->setShowToolTips(true);
   chart->layout()->setContentsMargins(0, 0, 0, 0);
+  chart->setMargins({20, 11, 11, 11});
 
   QToolButton *remove_btn = new QToolButton();
   remove_btn->setIcon(bootstrapPixmap("x"));
@@ -293,7 +300,8 @@ ChartView::ChartView(QWidget *parent) : QChartView(nullptr, parent) {
 
   setChart(chart);
   setRenderHint(QPainter::Antialiasing);
-  setRubberBand(QChartView::HorizontalRubberBand);
+  // TODO: enable zoomIn/seekTo in live streaming mode.
+  setRubberBand(can->liveStreaming() ? QChartView::NoRubberBand : QChartView::HorizontalRubberBand);
 
   QObject::connect(dbc(), &DBCManager::signalRemoved, this, &ChartView::signalRemoved);
   QObject::connect(dbc(), &DBCManager::signalUpdated, this, &ChartView::signalUpdated);
@@ -331,7 +339,6 @@ void ChartView::addSeries(const QString &msg_id, const Signal *sig) {
   sigs.push_back({.msg_id = msg_id, .address = address, .source = source, .sig = sig, .series = series});
   updateTitle();
   updateSeries(sig);
-  updateAxisY();
   emit seriesAdded(msg_id, sig);
 }
 
@@ -366,7 +373,6 @@ void ChartView::signalUpdated(const Signal *sig) {
     updateTitle();
     // TODO: don't update series if only name changed.
     updateSeries(sig);
-    updateAxisY();
   }
 }
 
@@ -448,22 +454,21 @@ void ChartView::setDisplayRange(double min, double max) {
   }
 }
 
-void ChartView::updateSeries(const Signal *sig) {
+void ChartView::updateSeries(const Signal *sig, const std::vector<Event*> *events, bool clear) {
-  auto events = can->events();
+  if (!events) events = can->events();
   if (!events || sigs.isEmpty()) return;
 
   for (auto &s : sigs) {
     if (!sig || s.sig == sig) {
-      s.vals.clear();
+      if (clear) {
-      s.vals.reserve((events_range.second - events_range.first) * 1000);  // [n]seconds * 1000hz
+        s.vals.clear();
-      s.min_y = std::numeric_limits<double>::max();
+        s.last_value_mono_time = 0;
-      s.max_y = std::numeric_limits<double>::lowest();
+      }
-
       double route_start_time = can->routeStartTime();
-      Event begin_event(cereal::Event::Which::INIT_DATA, (route_start_time + events_range.first) * 1e9);
+      uint64_t begin_ts = can->liveStreaming() ? s.last_value_mono_time : (route_start_time + events_range.first) * 1e9;
-      auto begin = std::lower_bound(events->begin(), events->end(), &begin_event, Event::lessThan());
+      Event begin_event(cereal::Event::Which::INIT_DATA, begin_ts);
-      double end_ns = (route_start_time + events_range.second) * 1e9;
+      auto begin = std::upper_bound(events->begin(), events->end(), &begin_event, Event::lessThan());
-
+      uint64_t end_ns = can->liveStreaming() ? events->back()->mono_time : (route_start_time + events_range.second) * 1e9;
       for (auto it = begin; it != events->end() && (*it)->mono_time <= end_ns; ++it) {
         if ((*it)->which == cereal::Event::Which::CAN) {
           for (const auto &c : (*it)->event.getCan()) {
@@ -472,14 +477,20 @@ void ChartView::updateSeries(const Signal *sig) {
               double value = get_raw_value((uint8_t *)dat.begin(), dat.size(), *s.sig);
               double ts = ((*it)->mono_time / (double)1e9) - route_start_time;  // seconds
               s.vals.push_back({ts, value});
-
-              if (value < s.min_y) s.min_y = value;
-              if (value > s.max_y) s.max_y = value;
             }
           }
         }
       }
+      if (!s.vals.isEmpty()) {
+        auto [min_v, max_v] = std::minmax_element(s.vals.begin(), s.vals.end(), [](auto &l, auto &r) { return l.y() < r.y(); });
+        s.min_y = min_v->y();
+        s.max_y = max_v->y();
+        s.last_value_mono_time = events->back()->mono_time;
+      } else {
+        s.min_y = s.max_y = 0;
+      }
       s.series->replace(s.vals);
+      updateAxisY();
     }
   }
 }
@@ -490,7 +501,7 @@ void ChartView::updateAxisY() {
 
   double min_y = std::numeric_limits<double>::max();
   double max_y = std::numeric_limits<double>::lowest();
-  if (events_range == std::pair{axis_x->min(), axis_x->max()}) {
+  if (can->liveStreaming() || events_range == std::pair{axis_x->min(), axis_x->max()}) {
     for (auto &s : sigs) {
       if (s.min_y < min_y) min_y = s.min_y;
       if (s.max_y > max_y) max_y = s.max_y;
@@ -583,7 +594,7 @@ void ChartView::mouseReleaseEvent(QMouseEvent *event) {
       emit zoomIn(min, max);
     }
     event->accept();
-  } else if (event->button() == Qt::RightButton) {
+  } else if (!can->liveStreaming() && event->button() == Qt::RightButton) {
     emit zoomReset();
     event->accept();
   } else {

tools/cabana/chartswidget.h

@@ -12,8 +12,8 @@
 #include <QtCharts/QLineSeries>
 #include <QtCharts/QValueAxis>
 
-#include "tools/cabana/canmessages.h"
 #include "tools/cabana/dbcmanager.h"
+#include "tools/cabana/streams/abstractstream.h"
 
 using namespace QtCharts;
 
@@ -25,7 +25,7 @@ public:
   void addSeries(const QString &msg_id, const Signal *sig);
   void removeSeries(const QString &msg_id, const Signal *sig);
   bool hasSeries(const QString &msg_id, const Signal *sig) const;
-  void updateSeries(const Signal *sig = nullptr);
+  void updateSeries(const Signal *sig = nullptr, const std::vector<Event*> *events = nullptr, bool clear = true);
   void setEventsRange(const std::pair<double, double> &range);
   void setDisplayRange(double min, double max);
   void setPlotAreaLeftPosition(int pos);
@@ -40,6 +40,7 @@ public:
     double min_y = 0;
     double max_y = 0;
     QVector<QPointF> vals;
+    uint64_t last_value_mono_time = 0;
   };
 
 signals:
@@ -128,7 +129,6 @@ private:
   QList<ChartView *> charts;
   uint32_t max_chart_range = 0;
   bool is_zoomed = false;
-  std::pair<double, double> event_range;
   std::pair<double, double> display_range;
   std::pair<double, double> zoomed_range;
   bool use_dark_theme = false;

tools/cabana/commands.h

@@ -2,7 +2,6 @@
 
 #include <QUndoCommand>
 
-#include "tools/cabana/canmessages.h"
 #include "tools/cabana/dbcmanager.h"
 
 class EditMsgCommand : public QUndoCommand {

tools/cabana/dbcmanager.h

@@ -28,9 +28,7 @@ public:
 
   static std::pair<uint8_t, uint32_t> parseId(const QString &id);
   inline static std::vector<std::string> allDBCNames() { return get_dbc_names(); }
-  inline std::map<uint32_t, DBCMsg> &allMsgs() { return msgs; }
   inline QString name() const { return dbc ? dbc->name.c_str() : ""; }
-
   void updateMsg(const QString &id, const QString &name, uint32_t size);
   void removeMsg(const QString &id);
   inline const std::map<uint32_t, DBCMsg> &messages() const { return msgs; }

tools/cabana/detailwidget.cc

@@ -8,9 +8,9 @@
 #include <QTimer>
 
 #include "selfdrive/ui/qt/util.h"
-#include "tools/cabana/canmessages.h"
 #include "tools/cabana/commands.h"
 #include "tools/cabana/dbcmanager.h"
+#include "tools/cabana/streams/abstractstream.h"
 
 // DetailWidget
 
@@ -99,7 +99,7 @@ DetailWidget::DetailWidget(ChartsWidget *charts, QWidget *parent) : charts(chart
   QObject::connect(binary_view, &BinaryView::resizeSignal, this, &DetailWidget::resizeSignal);
   QObject::connect(binary_view, &BinaryView::addSignal, this, &DetailWidget::addSignal);
   QObject::connect(tab_widget, &QTabWidget::currentChanged, [this]() { updateState(); });
-  QObject::connect(can, &CANMessages::msgsReceived, this, &DetailWidget::updateState);
+  QObject::connect(can, &AbstractStream::msgsReceived, this, &DetailWidget::updateState);
   QObject::connect(dbc(), &DBCManager::DBCFileChanged, [this]() { dbcMsgChanged(); });
   QObject::connect(tabbar, &QTabBar::customContextMenuRequested, this, &DetailWidget::showTabBarContextMenu);
   QObject::connect(tabbar, &QTabBar::currentChanged, [this](int index) {

tools/cabana/historylog.cc

@@ -222,8 +222,13 @@ LogsWidget::LogsWidget(QWidget *parent) : QWidget(parent) {
   QObject::connect(comp_box, SIGNAL(activated(int)), this, SLOT(setFilter()));
   QObject::connect(value_edit, &QLineEdit::textChanged, this, &LogsWidget::setFilter);
   QObject::connect(dynamic_mode, &QCheckBox::stateChanged, model, &HistoryLogModel::setDynamicMode);
-  QObject::connect(can, &CANMessages::seekedTo, model, &HistoryLogModel::refresh);
+  QObject::connect(can, &AbstractStream::seekedTo, model, &HistoryLogModel::refresh);
-  QObject::connect(can, &CANMessages::eventsMerged, model, &HistoryLogModel::segmentsMerged);
+  QObject::connect(can, &AbstractStream::eventsMerged, model, &HistoryLogModel::segmentsMerged);
+
+  if (can->liveStreaming()) {
+    dynamic_mode->setChecked(true);
+    dynamic_mode->setEnabled(false);
+  }
 }
 
 void LogsWidget::setMessage(const QString &message_id) {

tools/cabana/historylog.h

@@ -7,8 +7,8 @@
 #include <QLineEdit>
 #include <QTableView>
 
-#include "tools/cabana/canmessages.h"
 #include "tools/cabana/dbcmanager.h"
+#include "tools/cabana/streams/abstractstream.h"
 
 class HeaderView : public QHeaderView {
 public:

tools/cabana/mainwin.cc

@@ -58,8 +58,7 @@ MainWindow::MainWindow() : QMainWindow() {
   QObject::connect(this, &MainWindow::updateProgressBar, this, &MainWindow::updateDownloadProgress);
   QObject::connect(messages_widget, &MessagesWidget::msgSelectionChanged, detail_widget, &DetailWidget::setMessage);
   QObject::connect(charts_widget, &ChartsWidget::dock, this, &MainWindow::dockCharts);
-  QObject::connect(charts_widget, &ChartsWidget::rangeChanged, video_widget, &VideoWidget::rangeChanged);
+  QObject::connect(can, &AbstractStream::streamStarted, this, &MainWindow::loadDBCFromFingerprint);
-  QObject::connect(can, &CANMessages::streamStarted, this, &MainWindow::loadDBCFromFingerprint);
   QObject::connect(dbc(), &DBCManager::DBCFileChanged, this, &MainWindow::DBCFileChanged);
   QObject::connect(detail_widget->undo_stack, &QUndoStack::indexChanged, [this](int index) {
     setWindowTitle(tr("%1%2 - Cabana").arg(index > 0 ? "* " : "").arg(dbc()->name()));
@@ -123,9 +122,13 @@ void MainWindow::createDockWindows() {
   charts_layout->setContentsMargins(0, 0, 0, 0);
   charts_layout->addWidget(charts_widget);
 
-  video_widget = new VideoWidget(this);
   video_splitter = new QSplitter(Qt::Vertical,this);
-  video_splitter->addWidget(video_widget);
+
+  if (!can->liveStreaming()) {
+    video_widget = new VideoWidget(this);
+    video_splitter->addWidget(video_widget);
+    QObject::connect(charts_widget, &ChartsWidget::rangeChanged, video_widget, &VideoWidget::rangeChanged);
+  }
   video_splitter->addWidget(charts_container);
   video_splitter->restoreState(settings.video_splitter_state);
 

tools/cabana/mainwin.h

@@ -47,7 +47,7 @@ protected:
   void setOption();
   void findSimilarBits();
 
-  VideoWidget *video_widget;
+  VideoWidget *video_widget = nullptr;
   QDockWidget *video_dock;
   MessagesWidget *messages_widget;
   DetailWidget *detail_widget;

tools/cabana/messageswidget.cc

@@ -26,7 +26,6 @@ MessagesWidget::MessagesWidget(QWidget *parent) : QWidget(parent) {
   table_widget->setItemDelegateForColumn(4, new MessageBytesDelegate(table_widget));
   table_widget->setSelectionBehavior(QAbstractItemView::SelectRows);
   table_widget->setSelectionMode(QAbstractItemView::SingleSelection);
-  table_widget->setSizePolicy(QSizePolicy::Preferred, QSizePolicy::Expanding);
   table_widget->setSortingEnabled(true);
   table_widget->sortByColumn(0, Qt::AscendingOrder);
   table_widget->setColumnWidth(0, 250);
@@ -39,7 +38,7 @@ MessagesWidget::MessagesWidget(QWidget *parent) : QWidget(parent) {
 
   // signals/slots
   QObject::connect(filter, &QLineEdit::textChanged, model, &MessageListModel::setFilterString);
-  QObject::connect(can, &CANMessages::msgsReceived, model, &MessageListModel::msgsReceived);
+  QObject::connect(can, &AbstractStream::msgsReceived, model, &MessageListModel::msgsReceived);
   QObject::connect(dbc(), &DBCManager::DBCFileChanged, model, &MessageListModel::sortMessages);
   QObject::connect(dbc(), &DBCManager::msgUpdated, model, &MessageListModel::sortMessages);
   QObject::connect(dbc(), &DBCManager::msgRemoved, model, &MessageListModel::sortMessages);

tools/cabana/messageswidget.h

@@ -4,7 +4,7 @@
 #include <QTableView>
 #include <QStyledItemDelegate>
 
-#include "tools/cabana/canmessages.h"
+#include "tools/cabana/streams/abstractstream.h"
 
 class MessageListModel : public QAbstractTableModel {
 Q_OBJECT

tools/cabana/settings.cc

@@ -28,7 +28,7 @@ void Settings::save() {
 void Settings::load() {
   QSettings s("settings", QSettings::IniFormat);
   fps = s.value("fps", 10).toInt();
-  cached_segment_limit = s.value("cached_segment", 3).toInt();
+  cached_segment_limit = s.value("cached_segment", 5).toInt();
   chart_height = s.value("chart_height", 200).toInt();
   max_chart_x_range = s.value("max_chart_x_range", 3 * 60).toInt();
   chart_column_count = s.value("chart_column_count", 1).toInt();
@@ -51,13 +51,13 @@ SettingsDlg::SettingsDlg(QWidget *parent) : QDialog(parent) {
   form_layout->addRow("FPS", fps);
 
   cached_segment = new QSpinBox(this);
-  cached_segment->setRange(3, 60);
+  cached_segment->setRange(5, 60);
   cached_segment->setSingleStep(1);
   cached_segment->setValue(settings.cached_segment_limit);
   form_layout->addRow(tr("Cached segments limit"), cached_segment);
 
   max_chart_x_range = new QSpinBox(this);
-  max_chart_x_range->setRange(1, 60);
+  max_chart_x_range->setRange(3, 60);
   max_chart_x_range->setSingleStep(1);
   max_chart_x_range->setValue(settings.max_chart_x_range / 60);
   form_layout->addRow(tr("Chart range (minutes)"), max_chart_x_range);

tools/cabana/settings.h

@@ -14,7 +14,7 @@ public:
   void load();
 
   int fps = 10;
-  int cached_segment_limit = 3;
+  int cached_segment_limit = 5;
   int chart_height = 200;
   int chart_column_count = 1;
   int max_chart_x_range = 3 * 60; // 3 minutes

tools/cabana/signaledit.h

@@ -7,8 +7,8 @@
 #include <QToolButton>
 
 #include "selfdrive/ui/qt/widgets/controls.h"
-#include "tools/cabana/canmessages.h"
 #include "tools/cabana/dbcmanager.h"
+#include "tools/cabana/streams/abstractstream.h"
 
 class SignalForm : public QWidget {
   Q_OBJECT

tools/cabana/streams/abstractstream.cc

@@ -1,47 +1,13 @@
-#include "tools/cabana/canmessages.h"
+#include "tools/cabana/streams/abstractstream.h"
-#include "tools/cabana/dbcmanager.h"
 
-CANMessages *can = nullptr;
+AbstractStream *can = nullptr;
 
-CANMessages::CANMessages(QObject *parent) : QObject(parent) {
+AbstractStream::AbstractStream(QObject *parent, bool is_live_streaming) : is_live_streaming(is_live_streaming), QObject(parent) {
   can = this;
-  QObject::connect(this, &CANMessages::received, this, &CANMessages::process, Qt::QueuedConnection);
+  QObject::connect(this, &AbstractStream::received, this, &AbstractStream::process, Qt::QueuedConnection);
-  QObject::connect(&settings, &Settings::changed, this, &CANMessages::settingChanged);
 }
 
-CANMessages::~CANMessages() {
+void AbstractStream::process(QHash<QString, CanData> *messages) {
-  replay->stop();
-}
-
-static bool event_filter(const Event *e, void *opaque) {
-  CANMessages *c = (CANMessages *)opaque;
-  return c->eventFilter(e);
-}
-
-static QColor blend(QColor a, QColor b) {
-  return QColor((a.red() + b.red()) / 2, (a.green() + b.green()) / 2, (a.blue() + b.blue()) / 2, (a.alpha() + b.alpha()) / 2);
-}
-
-bool CANMessages::loadRoute(const QString &route, const QString &data_dir, uint32_t replay_flags) {
-  replay = new Replay(route, {"can", "roadEncodeIdx", "wideRoadEncodeIdx", "carParams"}, {}, nullptr, replay_flags, data_dir, this);
-  replay->setSegmentCacheLimit(settings.cached_segment_limit);
-  replay->installEventFilter(event_filter, this);
-  QObject::connect(replay, &Replay::seekedTo, this, &CANMessages::seekedTo);
-  QObject::connect(replay, &Replay::segmentsMerged, this, &CANMessages::eventsMerged);
-  QObject::connect(replay, &Replay::streamStarted, this, &CANMessages::streamStarted);
-  if (replay->load()) {
-    const auto &segments = replay->route()->segments();
-    if (std::none_of(segments.begin(), segments.end(), [](auto &s) { return s.second.rlog.length() > 0; })) {
-      qWarning() << "no rlogs in route" << route;
-      return false;
-    }
-    replay->start();
-    return true;
-  }
-  return false;
-}
-
-void CANMessages::process(QHash<QString, CanData> *messages) {
   for (auto it = messages->begin(); it != messages->end(); ++it) {
     can_msgs[it.key()] = it.value();
   }
@@ -51,7 +17,11 @@ void CANMessages::process(QHash<QString, CanData> *messages) {
   processing = false;
 }
 
-bool CANMessages::eventFilter(const Event *event) {
+static QColor blend(QColor a, QColor b) {
+  return QColor((a.red() + b.red()) / 2, (a.green() + b.green()) / 2, (a.blue() + b.blue()) / 2, (a.alpha() + b.alpha()) / 2);
+}
+
+bool AbstractStream::updateEvent(const Event *event) {
   static std::unique_ptr new_msgs = std::make_unique<QHash<QString, CanData>>();
   static QHash<QString, QByteArray> prev_dat;
   static QHash<QString, QList<QColor>> colors;
@@ -59,7 +29,7 @@ bool CANMessages::eventFilter(const Event *event) {
   static double prev_update_ts = 0;
 
   if (event->which == cereal::Event::Which::CAN) {
-    double current_sec = replay->currentSeconds();
+    double current_sec = currentSec();
     if (counters_begin_sec == 0 || counters_begin_sec >= current_sec) {
       new_msgs->clear();
       counters.clear();
@@ -79,7 +49,7 @@ bool CANMessages::eventFilter(const Event *event) {
         data.freq = data.count / delta;
       }
 
-      // Init colors
+       // Init colors
       if (colors[id].size() != data.dat.size()) {
         colors[id].clear();
         for (int i = 0; i < data.dat.size(); i++){
@@ -146,18 +116,3 @@ bool CANMessages::eventFilter(const Event *event) {
   }
   return true;
 }
-
-void CANMessages::seekTo(double ts) {
-  replay->seekTo(std::max(double(0), ts), false);
-  counters_begin_sec = 0;
-  emit updated();
-}
-
-void CANMessages::pause(bool pause) { 
-  replay->pause(pause); 
-  emit (pause ? paused() : resume());
-}
-
-void CANMessages::settingChanged() {
-  replay->setSegmentCacheLimit(settings.cached_segment_limit);
-}

tools/cabana/streams/abstractstream.h

@@ -0,0 +1,76 @@
+#pragma once
+
+#include <atomic>
+
+#include <QColor>
+#include <QHash>
+
+#include "tools/cabana/settings.h"
+#include "tools/replay/replay.h"
+
+struct CanData {
+  double ts = 0.;
+  uint8_t src = 0;
+  uint32_t address = 0;
+  uint32_t count = 0;
+  uint32_t freq = 0;
+  QByteArray dat;
+  QList<QColor> colors;
+};
+
+class AbstractStream : public QObject {
+  Q_OBJECT
+
+public:
+  AbstractStream(QObject *parent, bool is_live_streaming);
+  virtual ~AbstractStream() {};
+  inline bool liveStreaming() const { return is_live_streaming; }
+  virtual void seekTo(double ts) {}
+  virtual QString routeName() const = 0;
+  virtual QString carFingerprint() const { return ""; }
+  virtual double totalSeconds() const { return 0; }
+  virtual double routeStartTime() const { return 0; }
+  virtual double currentSec() const = 0;
+  virtual QDateTime currentDateTime() const { return {}; }
+  virtual const CanData &lastMessage(const QString &id) { return can_msgs[id]; }
+  virtual VisionStreamType visionStreamType() const { return VISION_STREAM_ROAD; }
+  virtual const Route *route() const { return nullptr; }
+  virtual const std::vector<Event *> *events() const = 0;
+  virtual void setSpeed(float speed) {}
+  virtual bool isPaused() const { return false; }
+  virtual void pause(bool pause) {}
+  virtual const std::vector<std::tuple<int, int, TimelineType>> getTimeline() { return {}; }
+
+signals:
+  void paused();
+  void resume();
+  void seekedTo(double sec);
+  void streamStarted();
+  void eventsMerged();
+  void updated();
+  void msgsReceived(const QHash<QString, CanData> *);
+  void received(QHash<QString, CanData> *);
+
+public:
+  QMap<QString, CanData> can_msgs;
+
+protected:
+  void process(QHash<QString, CanData> *);
+  bool updateEvent(const Event *event);
+
+  bool is_live_streaming = false;
+  std::atomic<double> counters_begin_sec = 0;
+  std::atomic<bool> processing = false;
+  QHash<QString, uint32_t> counters;
+};
+
+inline QString toHex(const QByteArray &dat) { return dat.toHex(' ').toUpper(); }
+inline char toHex(uint value) { return "0123456789ABCDEF"[value & 0xF]; }
+inline const QString &getColor(int i) {
+  // TODO: add more colors
+  static const QString SIGNAL_COLORS[] = {"#9FE2BF", "#40E0D0", "#6495ED", "#CCCCFF", "#FF7F50", "#FFBF00"};
+  return SIGNAL_COLORS[i % std::size(SIGNAL_COLORS)];
+}
+
+// A global pointer referring to the unique AbstractStream object
+extern AbstractStream *can;

tools/cabana/streams/livestream.cc

@@ -0,0 +1,71 @@
+#include "tools/cabana/streams/livestream.h"
+
+LiveStream::LiveStream(QObject *parent, QString address) : zmq_address(address), AbstractStream(parent, true) {
+  if (!zmq_address.isEmpty()) {
+    setenv("ZMQ", "1", 1);
+  }
+  updateCachedNS();
+  QObject::connect(&settings, &Settings::changed, this, &LiveStream::updateCachedNS);
+  stream_thread = new QThread(this);
+  QObject::connect(stream_thread, &QThread::started, [=]() { streamThread(); });
+  QObject::connect(stream_thread, &QThread::finished, stream_thread, &QThread::deleteLater);
+  stream_thread->start();
+}
+
+LiveStream::~LiveStream() {
+  stream_thread->requestInterruption();
+  stream_thread->quit();
+  stream_thread->wait();
+  for (Event *e : can_events) ::delete e;
+  for (auto m : messages) delete m;
+}
+
+void LiveStream::streamThread() {
+  std::unique_ptr<Context> context(Context::create());
+  std::string address = zmq_address.isEmpty() ? "127.0.0.1" : zmq_address.toStdString();
+  std::unique_ptr<SubSocket> sock(SubSocket::create(context.get(), "can", address));
+  assert(sock != NULL);
+  sock->setTimeout(50);
+
+  // run as fast as messages come in
+  while (!QThread::currentThread()->isInterruptionRequested()) {
+    Message *msg = sock->receive(true);
+    if (!msg) {
+      QThread::msleep(50);
+      continue;
+    }
+    AlignedBuffer *buf = messages.emplace_back(new AlignedBuffer());
+    Event *evt = ::new Event(buf->align(msg));
+    delete msg;
+
+    {
+      std::lock_guard lk(lock);
+      can_events.push_back(evt);
+      if ((evt->mono_time - can_events.front()->mono_time) > cache_ns) {
+        ::delete can_events.front();
+        delete messages.front();
+        can_events.pop_front();
+        messages.pop_front();
+      }
+    }
+    if (start_ts == 0) {
+      start_ts = evt->mono_time;
+      emit streamStarted();
+    }
+    current_ts = evt->mono_time;
+    if (start_ts > current_ts) {
+      qDebug() << "stream is looping back to old time stamp";
+      start_ts = current_ts.load();
+    }
+    updateEvent(evt);
+    // TODO: write stream to log file to replay it with cabana --data_dir flag.
+  }
+}
+
+const std::vector<Event *> *LiveStream::events() const {
+  std::lock_guard lk(lock);
+  events_vector.clear();
+  events_vector.reserve(can_events.size());
+  std::copy(can_events.begin(), can_events.end(), std::back_inserter(events_vector));
+  return &events_vector;
+}

tools/cabana/streams/livestream.h

@@ -0,0 +1,31 @@
+#pragma once
+
+#include "tools/cabana/streams/abstractstream.h"
+
+class LiveStream : public AbstractStream {
+  Q_OBJECT
+
+public:
+  LiveStream(QObject *parent, QString address = {});
+  ~LiveStream();
+  inline QString routeName() const override {
+    return QString("Live Streaming From %1").arg(zmq_address.isEmpty() ? "127.0.0.1" : zmq_address);
+  }
+  inline double routeStartTime() const override { return start_ts / (double)1e9; }
+  inline double currentSec() const override { return (current_ts - start_ts) / (double)1e9; }
+  const std::vector<Event *> *events() const override;
+
+protected:
+  void streamThread();
+  void updateCachedNS() { cache_ns = (settings.cached_segment_limit * 60) * 1e9; }
+
+  mutable std::mutex lock;
+  mutable std::vector<Event *> events_vector;
+  std::deque<Event *> can_events;
+  std::deque<AlignedBuffer *> messages;
+  std::atomic<uint64_t> start_ts = 0;
+  std::atomic<uint64_t> current_ts = 0;
+  std::atomic<uint64_t> cache_ns = 0;
+  const QString zmq_address;
+  QThread *stream_thread;
+};

tools/cabana/streams/replaystream.cc

@@ -0,0 +1,54 @@
+#include "tools/cabana/streams/replaystream.h"
+
+#include "tools/cabana/dbcmanager.h"
+
+ReplayStream::ReplayStream(QObject *parent) : AbstractStream(parent, false) {
+  QObject::connect(&settings, &Settings::changed, [this]() {
+    if (replay) replay->setSegmentCacheLimit(settings.cached_segment_limit);
+  });
+}
+
+ReplayStream::~ReplayStream() {
+  if (replay) replay->stop();
+}
+
+static bool event_filter(const Event *e, void *opaque) {
+  return ((ReplayStream *)opaque)->eventFilter(e);
+}
+
+bool ReplayStream::loadRoute(const QString &route, const QString &data_dir, uint32_t replay_flags) {
+  replay = new Replay(route, {"can", "roadEncodeIdx", "wideRoadEncodeIdx", "carParams"}, {}, nullptr, replay_flags, data_dir, this);
+  replay->setSegmentCacheLimit(settings.cached_segment_limit);
+  replay->installEventFilter(event_filter, this);
+  QObject::connect(replay, &Replay::seekedTo, this, &AbstractStream::seekedTo);
+  QObject::connect(replay, &Replay::segmentsMerged, this, &AbstractStream::eventsMerged);
+  QObject::connect(replay, &Replay::streamStarted, this, &AbstractStream::streamStarted);
+  if (replay->load()) {
+    const auto &segments = replay->route()->segments();
+    if (std::none_of(segments.begin(), segments.end(), [](auto &s) { return s.second.rlog.length() > 0; })) {
+      qWarning() << "no rlogs in route" << route;
+      return false;
+    }
+    replay->start();
+    return true;
+  }
+  return false;
+}
+
+bool ReplayStream::eventFilter(const Event *event) {
+  if (event->which == cereal::Event::Which::CAN) {
+    updateEvent(event);
+  }
+  return true;
+}
+
+void ReplayStream::seekTo(double ts) {
+  replay->seekTo(std::max(double(0), ts), false);
+  counters_begin_sec = 0;
+  emit updated();
+}
+
+void ReplayStream::pause(bool pause) {
+  replay->pause(pause);
+  emit(pause ? paused() : resume());
+}

tools/cabana/streams/replaystream.h

@@ -0,0 +1,32 @@
+#pragma once
+
+#include "opendbc/can/common_dbc.h"
+#include "tools/cabana/streams/abstractstream.h"
+#include "tools/cabana/settings.h"
+
+class ReplayStream : public AbstractStream {
+  Q_OBJECT
+
+public:
+  ReplayStream(QObject *parent);
+  ~ReplayStream();
+  bool loadRoute(const QString &route, const QString &data_dir, uint32_t replay_flags = REPLAY_FLAG_NONE);
+  bool eventFilter(const Event *event);
+  void seekTo(double ts) override;
+  inline QString routeName() const override { return replay->route()->name(); }
+  inline QString carFingerprint() const override { return replay->carFingerprint().c_str(); }
+  inline VisionStreamType visionStreamType() const override { return replay->hasFlag(REPLAY_FLAG_ECAM) ? VISION_STREAM_WIDE_ROAD : VISION_STREAM_ROAD; }
+  inline double totalSeconds() const override { return replay->totalSeconds(); }
+  inline double routeStartTime() const override { return replay->routeStartTime() / (double)1e9; }
+  inline double currentSec() const override { return replay->currentSeconds(); }
+  inline QDateTime currentDateTime() const override { return replay->currentDateTime(); }
+  inline const Route *route() const override { return replay->route(); }
+  inline const std::vector<Event *> *events() const override { return replay->events(); }
+  inline void setSpeed(float speed) override { replay->setSpeed(speed); }
+  inline bool isPaused() const override { return replay->isPaused(); }
+  void pause(bool pause) override;
+  inline const std::vector<std::tuple<int, int, TimelineType>> getTimeline() override { return replay->getTimeline(); }
+
+private:
+  Replay *replay = nullptr;
+};

tools/cabana/tools/findsimilarbits.cc

@@ -7,8 +7,8 @@
 #include <QPushButton>
 #include <QRadioButton>
 
-#include "tools/cabana/canmessages.h"
 #include "tools/cabana/dbcmanager.h"
+#include "tools/cabana/streams/abstractstream.h"
 
 FindSimilarBitsDlg::FindSimilarBitsDlg(QWidget *parent) : QDialog(parent, Qt::WindowFlags() | Qt::Window) {
   setWindowTitle(tr("Find similar bits"));

tools/cabana/videowidget.cc

@@ -70,10 +70,10 @@ VideoWidget::VideoWidget(QWidget *parent) : QWidget(parent) {
   QObject::connect(slider, &QSlider::valueChanged, [=](int value) { time_label->setText(formatTime(value / 1000)); });
   QObject::connect(cam_widget, &CameraWidget::clicked, []() { can->pause(!can->isPaused()); });
   QObject::connect(play_btn, &QPushButton::clicked, []() { can->pause(!can->isPaused()); });
-  QObject::connect(can, &CANMessages::updated, this, &VideoWidget::updateState);
+  QObject::connect(can, &AbstractStream::updated, this, &VideoWidget::updateState);
-  QObject::connect(can, &CANMessages::paused, this, &VideoWidget::updatePlayBtnState);
+  QObject::connect(can, &AbstractStream::paused, this, &VideoWidget::updatePlayBtnState);
-  QObject::connect(can, &CANMessages::resume, this, &VideoWidget::updatePlayBtnState);
+  QObject::connect(can, &AbstractStream::resume, this, &VideoWidget::updatePlayBtnState);
-  QObject::connect(can, &CANMessages::streamStarted, [this]() {
+  QObject::connect(can, &AbstractStream::streamStarted, [this]() {
     end_time_label->setText(formatTime(can->totalSeconds()));
     slider->setRange(0, can->totalSeconds() * 1000);
   });
@@ -130,7 +130,7 @@ Slider::Slider(QWidget *parent) : QSlider(Qt::Horizontal, parent) {
   setMouseTracking(true);
 
   QObject::connect(can, SIGNAL(streamStarted()), timer, SLOT(start()));
-  QObject::connect(can, &CANMessages::streamStarted, this, &Slider::streamStarted);
+  QObject::connect(can, &AbstractStream::streamStarted, this, &Slider::streamStarted);
 }
 
 void Slider::streamStarted() {

tools/cabana/videowidget.h

@@ -11,7 +11,7 @@
 
 #include "selfdrive/ui/qt/widgets/cameraview.h"
 #include "selfdrive/ui/qt/widgets/controls.h"
-#include "tools/cabana/canmessages.h"
+#include "tools/cabana/streams/abstractstream.h"
 
 class Slider : public QSlider {
   Q_OBJECT