mawei/dragonpilot
1
0
Fork 0
mirror of https://github.com/dragonpilot/dragonpilot.git synced 2026-07-08 22:42:05 +08:00
Code Issues Packages Projects Releases Wiki Activity

Cabana: support live streaming (#26946)

Browse Source 
* 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
This commit is contained in:
Dean Lee
2023-01-20 07:26:00 +08:00
committed by GitHub
parent fd4dc109e1
commit f9490739ab
27 changed files with 395 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/cabana/SConscript
+2 -2
View File
@@ -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',
'canmessages.cc', 'commands.cc', 'messageswidget.cc', 'settings.cc', 'detailwidget.cc', 'tools/findsimilarbits.cc'], LIBS=cabana_libs, FRAMEWORKS=base_frameworks)
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',
'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
+1 -1
View File
@@ -6,7 +6,7 @@
#include <QPainter>
#include <QToolTip>
#include "tools/cabana/canmessages.h"
#include "tools/cabana/streams/abstractstream.h"
// BinaryView
tools/cabana/cabana.cc
+28 -19
View File
@@ -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();
uint32_t replay_flags = REPLAY_FLAG_NONE;
if (cmd_parser.isSet("ecam")) {
replay_flags |= REPLAY_FLAG_ECAM;
} else if (cmd_parser.isSet("qcam")) {
replay_flags |= REPLAY_FLAG_QCAMERA;
}
// TODO: Remove when OpenpilotPrefix supports ZMQ
if (cmd_parser.isSet("stream")) {
stream.reset(new LiveStream(&app, cmd_parser.value("zmq")));
} else {
// TODO: Remove when OpenpilotPrefix supports ZMQ
#ifndef __APPLE__
OpenpilotPrefix op_prefix;
op_prefix.reset(new OpenpilotPrefix());
#endif
CANMessages p(&app);
int ret = 0;
if (p.loadRoute(route, cmd_parser.value("data_dir"), replay_flags)) {
MainWindow w;
w.show();
ret = app.exec();
const QString route = args.empty() ? DEMO_ROUTE : args.first();
uint32_t replay_flags = REPLAY_FLAG_NONE;
if (cmd_parser.isSet("ecam")) {
replay_flags |= REPLAY_FLAG_ECAM;
} else if (cmd_parser.isSet("qcam")) {
replay_flags |= REPLAY_FLAG_QCAMERA;
}
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
-86
View File
@@ -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
+53 -42
View File
@@ -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, &CANMessages::updated, this, &ChartsWidget::updateState);
QObject::connect(can, &AbstractStream::eventsMerged, 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;
double current_sec = can->currentSec();
if (current_sec < display_range.first || current_sec >= (display_range.second - 5)) {
// reached the end, or seeked to a timestamp out of range.
display_range.first = current_sec - 5;
}
display_range.first = std::floor(std::max(display_range.first, event_range.first) * 10.0) / 10.0;
display_range.second = std::floor(std::min(display_range.first + max_chart_range, event_range.second) * 10.0) / 10.0;
if (prev_range != display_range) {
QFutureSynchronizer<void> future_synchronizer;
for (auto c : charts)
future_synchronizer.addFuture(QtConcurrent::run(c, &ChartView::setEventsRange, display_range));
auto events = can->events();
double min_event_sec = (events->front()->mono_time / (double)1e9) - can->routeStartTime();
double max_event_sec = (events->back()->mono_time / (double)1e9) - can->routeStartTime();
const double cur_sec = can->currentSec();
if (!can->liveStreaming()) {
auto prev_range = display_range;
if (cur_sec < display_range.first || cur_sec >= (display_range.second - 5)) {
// reached the end, or seeked to a timestamp out of range.
display_range.first = cur_sec - 5;
}
display_range.first = std::floor(std::max(min_event_sec, display_range.first));
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) {
auto events = can->events();
void ChartView::updateSeries(const Signal *sig, const std::vector<Event*> *events, bool clear) {
if (!events) events = can->events();
if (!events || sigs.isEmpty()) return;
for (auto &s : sigs) {
if (!sig || s.sig == sig) {
s.vals.clear();
s.vals.reserve((events_range.second - events_range.first) * 1000); // [n]seconds * 1000hz
s.min_y = std::numeric_limits<double>::max();
s.max_y = std::numeric_limits<double>::lowest();
if (clear) {
s.vals.clear();
s.last_value_mono_time = 0;
}
double route_start_time = can->routeStartTime();
Event begin_event(cereal::Event::Which::INIT_DATA, (route_start_time + events_range.first) * 1e9);
auto begin = std::lower_bound(events->begin(), events->end(), &begin_event, Event::lessThan());
double end_ns = (route_start_time + events_range.second) * 1e9;
uint64_t begin_ts = can->liveStreaming() ? s.last_value_mono_time : (route_start_time + events_range.first) * 1e9;
Event begin_event(cereal::Event::Which::INIT_DATA, begin_ts);
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
+3 -3
View File
@@ -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
-1
View File
@@ -2,7 +2,6 @@
#include <QUndoCommand>
#include "tools/cabana/canmessages.h"
#include "tools/cabana/dbcmanager.h"
class EditMsgCommand : public QUndoCommand {
tools/cabana/dbcmanager.h
-2
View File
@@ -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
+2 -2
View File
@@ -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
+7 -2
View File
@@ -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, &CANMessages::eventsMerged, model, &HistoryLogModel::segmentsMerged);
QObject::connect(can, &AbstractStream::seekedTo, model, &HistoryLogModel::refresh);
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
+1 -1
View File
@@ -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
+7 -4
View File
@@ -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, &CANMessages::streamStarted, this, &MainWindow::loadDBCFromFingerprint);
QObject::connect(can, &AbstractStream::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
+1 -1
View File
@@ -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
+1 -2
View File
@@ -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
+1 -1
View File
@@ -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
+3 -3
View File
@@ -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
+1 -1
View File
@@ -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
+1 -1
View File
@@ -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/canmessages.cc → tools/cabana/streams/abstractstream.cc
+12 -57
View File
@@ -1,47 +1,13 @@
#include "tools/cabana/canmessages.h"
#include "tools/cabana/dbcmanager.h"
#include "tools/cabana/streams/abstractstream.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(&settings, &Settings::changed, this, &CANMessages::settingChanged);
QObject::connect(this, &AbstractStream::received, this, &AbstractStream::process, Qt::QueuedConnection);
}
CANMessages::~CANMessages() {
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) {
void AbstractStream::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
+76
View File
@@ -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
+71
View File
@@ -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
+31
View File
@@ -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
+54
View File
@@ -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
+32
View File
@@ -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
+1 -1
View File
@@ -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
+5 -5
View File
@@ -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, &CANMessages::paused, this, &VideoWidget::updatePlayBtnState);
QObject::connect(can, &CANMessages::resume, this, &VideoWidget::updatePlayBtnState);
QObject::connect(can, &CANMessages::streamStarted, [this]() {
QObject::connect(can, &AbstractStream::updated, this, &VideoWidget::updateState);
QObject::connect(can, &AbstractStream::paused, this, &VideoWidget::updatePlayBtnState);
QObject::connect(can, &AbstractStream::resume, this, &VideoWidget::updatePlayBtnState);
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
+1 -1
View File
@@ -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