From e7cc70f3fa8345662be574e4b0aae4c0a5662b05 Mon Sep 17 00:00:00 2001 From: Adeeb Shihadeh Date: Sat, 28 Feb 2026 21:09:02 -0800 Subject: [PATCH] consolidate file downloading from C++ to Python (#37497) --- system/loggerd/SConscript | 2 +- tools/cabana/SConscript | 4 +- tools/cabana/mainwin.cc | 2 + tools/cabana/streams/routes.cc | 91 +++++++----- tools/cabana/streams/routes.h | 9 +- tools/cabana/utils/api.cc | 171 ----------------------- tools/cabana/utils/api.h | 47 ------- tools/lib/file_downloader.py | 148 ++++++++++++++++++++ tools/replay/SConscript | 4 +- tools/replay/api.cc | 162 ---------------------- tools/replay/api.h | 15 -- tools/replay/consoleui.cc | 1 + tools/replay/filereader.cc | 47 +------ tools/replay/filereader.h | 8 +- tools/replay/framereader.cc | 15 +- tools/replay/framereader.h | 4 +- tools/replay/logreader.cc | 4 +- tools/replay/logreader.h | 2 +- tools/replay/main.cc | 5 + tools/replay/py_downloader.cc | 218 +++++++++++++++++++++++++++++ tools/replay/py_downloader.h | 24 ++++ tools/replay/route.cc | 75 +++++----- tools/replay/route.h | 5 +- tools/replay/tests/test_replay.cc | 1 + tools/replay/timeline.cc | 2 +- tools/replay/util.cc | 221 ------------------------------ tools/replay/util.h | 6 - 27 files changed, 522 insertions(+), 771 deletions(-) delete mode 100644 tools/cabana/utils/api.cc delete mode 100644 tools/cabana/utils/api.h create mode 100755 tools/lib/file_downloader.py delete mode 100644 tools/replay/api.cc delete mode 100644 tools/replay/api.h create mode 100644 tools/replay/py_downloader.cc create mode 100644 tools/replay/py_downloader.h diff --git a/system/loggerd/SConscript b/system/loggerd/SConscript index fecf44885..827c1ce5d 100644 --- a/system/loggerd/SConscript +++ b/system/loggerd/SConscript @@ -20,4 +20,4 @@ env.Program('encoderd', ['encoderd.cc'], LIBS=libs + ["jpeg"]) env.Program('bootlog.cc', LIBS=libs) if GetOption('extras'): - env.Program('tests/test_logger', ['tests/test_runner.cc', 'tests/test_logger.cc', 'tests/test_zstd_writer.cc'], LIBS=libs + ['curl', 'crypto']) + env.Program('tests/test_logger', ['tests/test_runner.cc', 'tests/test_logger.cc', 'tests/test_zstd_writer.cc'], LIBS=libs) diff --git a/tools/cabana/SConscript b/tools/cabana/SConscript index e172278d9..cecb5ed8d 100644 --- a/tools/cabana/SConscript +++ b/tools/cabana/SConscript @@ -81,7 +81,7 @@ if arch == "Darwin": cabana_env['CPPPATH'] += [f"{brew_prefix}/include"] cabana_env['LIBPATH'] += [f"{brew_prefix}/lib"] -cabana_libs = [cereal, messaging, visionipc, replay_lib, 'avformat', 'avcodec', 'swresample', 'avutil', 'x264', 'z', 'bz2', 'zstd', 'curl', 'yuv', 'usb-1.0'] + qt_libs +cabana_libs = [cereal, messaging, visionipc, replay_lib, 'avformat', 'avcodec', 'swresample', 'avutil', 'x264', 'z', 'bz2', 'zstd', 'yuv', 'usb-1.0'] + qt_libs opendbc_path = '-DOPENDBC_FILE_PATH=\'"%s"\'' % (cabana_env.Dir("../../opendbc/dbc").abspath) cabana_env['CXXFLAGS'] += [opendbc_path] @@ -93,7 +93,7 @@ cabana_env.Depends(assets, Glob('/assets/*', exclude=[assets, assets_src, "asset cabana_lib = cabana_env.Library("cabana_lib", ['mainwin.cc', 'streams/socketcanstream.cc', 'streams/pandastream.cc', 'streams/devicestream.cc', 'streams/livestream.cc', 'streams/abstractstream.cc', 'streams/replaystream.cc', 'binaryview.cc', 'historylog.cc', 'videowidget.cc', 'signalview.cc', 'streams/routes.cc', 'dbc/dbc.cc', 'dbc/dbcfile.cc', 'dbc/dbcmanager.cc', - 'utils/export.cc', 'utils/util.cc', 'utils/elidedlabel.cc', 'utils/api.cc', + 'utils/export.cc', 'utils/util.cc', 'utils/elidedlabel.cc', 'chart/chartswidget.cc', 'chart/chart.cc', 'chart/signalselector.cc', 'chart/tiplabel.cc', 'chart/sparkline.cc', 'commands.cc', 'messageswidget.cc', 'streamselector.cc', 'settings.cc', 'panda.cc', 'cameraview.cc', 'detailwidget.cc', 'tools/findsimilarbits.cc', 'tools/findsignal.cc', 'tools/routeinfo.cc'], LIBS=cabana_libs, FRAMEWORKS=base_frameworks) diff --git a/tools/cabana/mainwin.cc b/tools/cabana/mainwin.cc index 1ea3733ed..a7040f891 100644 --- a/tools/cabana/mainwin.cc +++ b/tools/cabana/mainwin.cc @@ -24,6 +24,8 @@ #include "tools/cabana/streamselector.h" #include "tools/cabana/tools/findsignal.h" #include "tools/cabana/utils/export.h" +#include "tools/replay/py_downloader.h" +#include "tools/replay/util.h" MainWindow::MainWindow(AbstractStream *stream, const QString &dbc_file) : QMainWindow() { loadFingerprints(); diff --git a/tools/cabana/streams/routes.cc b/tools/cabana/streams/routes.cc index 5915c9806..8539a00b5 100644 --- a/tools/cabana/streams/routes.cc +++ b/tools/cabana/streams/routes.cc @@ -1,27 +1,33 @@ #include "tools/cabana/streams/routes.h" +#include #include #include #include #include #include +#include #include #include #include +#include +#include -class OneShotHttpRequest : public HttpRequest { -public: - OneShotHttpRequest(QObject *parent) : HttpRequest(parent, false) {} - void send(const QString &url) { - if (reply) { - reply->disconnect(); - reply->abort(); - reply->deleteLater(); - reply = nullptr; - } - sendRequest(url); +#include "tools/replay/py_downloader.h" + +namespace { + +// Parse a PyDownloader JSON response into (success, error_code). +std::pair checkApiResponse(const std::string &result) { + if (result.empty()) return {false, 500}; + auto doc = QJsonDocument::fromJson(QByteArray::fromStdString(result)); + if (doc.isObject() && doc.object().contains("error")) { + return {false, doc.object()["error"].toString() == "unauthorized" ? 401 : 500}; } -}; + return {true, 0}; +} + +} // namespace // The RouteListWidget class extends QListWidget to display a custom message when empty class RouteListWidget : public QListWidget { @@ -41,7 +47,7 @@ public: QString empty_text_ = tr("No items"); }; -RoutesDialog::RoutesDialog(QWidget *parent) : QDialog(parent), route_requester_(new OneShotHttpRequest(this)) { +RoutesDialog::RoutesDialog(QWidget *parent) : QDialog(parent) { setWindowTitle(tr("Remote routes")); QFormLayout *layout = new QFormLayout(this); @@ -52,41 +58,40 @@ RoutesDialog::RoutesDialog(QWidget *parent) : QDialog(parent), route_requester_( layout->addRow(button_box); device_list_->addItem(tr("Loading...")); - // Populate period selector with predefined durations period_selector_->addItem(tr("Last week"), 7); period_selector_->addItem(tr("Last 2 weeks"), 14); period_selector_->addItem(tr("Last month"), 30); period_selector_->addItem(tr("Last 6 months"), 180); period_selector_->addItem(tr("Preserved"), -1); - // Connect signals and slots - QObject::connect(route_requester_, &HttpRequest::requestDone, this, &RoutesDialog::parseRouteList); connect(device_list_, QOverload::of(&QComboBox::currentIndexChanged), this, &RoutesDialog::fetchRoutes); connect(period_selector_, QOverload::of(&QComboBox::currentIndexChanged), this, &RoutesDialog::fetchRoutes); connect(route_list_, &QListWidget::itemDoubleClicked, this, &QDialog::accept); - QObject::connect(button_box, &QDialogButtonBox::accepted, this, &QDialog::accept); - QObject::connect(button_box, &QDialogButtonBox::rejected, this, &QDialog::reject); + connect(button_box, &QDialogButtonBox::accepted, this, &QDialog::accept); + connect(button_box, &QDialogButtonBox::rejected, this, &QDialog::reject); - // Send request to fetch devices - HttpRequest *http = new HttpRequest(this, false); - QObject::connect(http, &HttpRequest::requestDone, this, &RoutesDialog::parseDeviceList); - http->sendRequest(CommaApi::BASE_URL + "/v1/me/devices/"); + // Fetch devices + QPointer self = this; + QtConcurrent::run([self]() { + std::string result = PyDownloader::getDevices(); + auto [success, error_code] = checkApiResponse(result); + QMetaObject::invokeMethod(qApp, [self, r = QString::fromStdString(result), success, error_code]() { + if (self) self->parseDeviceList(r, success, error_code); + }, Qt::QueuedConnection); + }); } -void RoutesDialog::parseDeviceList(const QString &json, bool success, QNetworkReply::NetworkError err) { +void RoutesDialog::parseDeviceList(const QString &json, bool success, int error_code) { if (success) { device_list_->clear(); - auto devices = QJsonDocument::fromJson(json.toUtf8()).array(); - for (const QJsonValue &device : devices) { + for (const QJsonValue &device : QJsonDocument::fromJson(json.toUtf8()).array()) { QString dongle_id = device["dongle_id"].toString(); device_list_->addItem(dongle_id, dongle_id); } } else { - bool unauthorized = (err == QNetworkReply::ContentAccessDenied || err == QNetworkReply::AuthenticationRequiredError); - QMessageBox::warning(this, tr("Error"), unauthorized ? tr("Unauthorized, Authenticate with tools/lib/auth.py") : tr("Network error")); + QMessageBox::warning(this, tr("Error"), error_code == 401 ? tr("Unauthorized. Authenticate with tools/lib/auth.py") : tr("Network error")); reject(); } - sender()->deleteLater(); } void RoutesDialog::fetchRoutes() { @@ -95,21 +100,31 @@ void RoutesDialog::fetchRoutes() { route_list_->clear(); route_list_->setEmptyText(tr("Loading...")); - // Construct URL with selected device and date range - QString url = QString("%1/v1/devices/%2").arg(CommaApi::BASE_URL, device_list_->currentText()); + + std::string did = device_list_->currentText().toStdString(); int period = period_selector_->currentData().toInt(); - if (period == -1) { - url += "/routes/preserved"; - } else { + + bool preserved = (period == -1); + int64_t start_ms = 0, end_ms = 0; + if (!preserved) { QDateTime now = QDateTime::currentDateTime(); - url += QString("/routes_segments?start=%1&end=%2") - .arg(now.addDays(-period).toMSecsSinceEpoch()) - .arg(now.toMSecsSinceEpoch()); + start_ms = now.addDays(-period).toMSecsSinceEpoch(); + end_ms = now.toMSecsSinceEpoch(); } - route_requester_->send(url); + + int request_id = ++fetch_id_; + QPointer self = this; + QtConcurrent::run([self, did, start_ms, end_ms, preserved, request_id]() { + std::string result = PyDownloader::getDeviceRoutes(did, start_ms, end_ms, preserved); + if (!self || self->fetch_id_ != request_id) return; + auto [success, error_code] = checkApiResponse(result); + QMetaObject::invokeMethod(qApp, [self, r = QString::fromStdString(result), success, error_code, request_id]() { + if (self && self->fetch_id_ == request_id) self->parseRouteList(r, success, error_code); + }, Qt::QueuedConnection); + }); } -void RoutesDialog::parseRouteList(const QString &json, bool success, QNetworkReply::NetworkError err) { +void RoutesDialog::parseRouteList(const QString &json, bool success, int error_code) { if (success) { for (const QJsonValue &route : QJsonDocument::fromJson(json.toUtf8()).array()) { QDateTime from, to; diff --git a/tools/cabana/streams/routes.h b/tools/cabana/streams/routes.h index 045dc6722..99fa67ef8 100644 --- a/tools/cabana/streams/routes.h +++ b/tools/cabana/streams/routes.h @@ -1,11 +1,10 @@ #pragma once +#include #include #include -#include "tools/cabana/utils/api.h" class RouteListWidget; -class OneShotHttpRequest; class RoutesDialog : public QDialog { Q_OBJECT @@ -14,12 +13,12 @@ public: QString route(); protected: - void parseDeviceList(const QString &json, bool success, QNetworkReply::NetworkError err); - void parseRouteList(const QString &json, bool success, QNetworkReply::NetworkError err); + void parseDeviceList(const QString &json, bool success, int error_code); + void parseRouteList(const QString &json, bool success, int error_code); void fetchRoutes(); QComboBox *device_list_; QComboBox *period_selector_; RouteListWidget *route_list_; - OneShotHttpRequest *route_requester_; + std::atomic fetch_id_{0}; }; diff --git a/tools/cabana/utils/api.cc b/tools/cabana/utils/api.cc deleted file mode 100644 index 89379a843..000000000 --- a/tools/cabana/utils/api.cc +++ /dev/null @@ -1,171 +0,0 @@ -#include "tools/cabana/utils/api.h" - -#include -#include - -#include -#include -#include -#include -#include -#include - -#include -#include - -#include "common/params.h" -#include "common/util.h" -#include "system/hardware/hw.h" -#include "tools/cabana/utils/util.h" - -QString getVersion() { - static QString version = QString::fromStdString(Params().get("Version")); - return version; -} - -QString getUserAgent() { - return "openpilot-" + getVersion(); -} - -std::optional getDongleId() { - std::string id = Params().get("DongleId"); - - if (!id.empty() && (id != "UnregisteredDevice")) { - return QString::fromStdString(id); - } else { - return {}; - } -} - -namespace CommaApi { - -EVP_PKEY *get_private_key() { - static std::unique_ptr pkey(nullptr, EVP_PKEY_free); - if (!pkey) { - FILE *fp = fopen(Path::rsa_file().c_str(), "rb"); - if (!fp) { - qDebug() << "No private key found, please run manager.py or registration.py"; - return nullptr; - } - pkey.reset(PEM_read_PrivateKey(fp, nullptr, nullptr, nullptr)); - fclose(fp); - } - return pkey.get(); -} - -QByteArray rsa_sign(const QByteArray &data) { - EVP_PKEY *pkey = get_private_key(); - if (!pkey) return {}; - - EVP_MD_CTX *mdctx = EVP_MD_CTX_new(); - if (!mdctx) return {}; - - QByteArray sig(EVP_PKEY_size(pkey), Qt::Uninitialized); - size_t sig_len = sig.size(); - - int ret = EVP_DigestSignInit(mdctx, nullptr, EVP_sha256(), nullptr, pkey); - ret &= EVP_DigestSignUpdate(mdctx, data.data(), data.size()); - ret &= EVP_DigestSignFinal(mdctx, (unsigned char*)sig.data(), &sig_len); - - EVP_MD_CTX_free(mdctx); - - if (ret != 1) return {}; - sig.resize(sig_len); - return sig; -} - -QString create_jwt(const QJsonObject &payloads, int expiry) { - QJsonObject header = {{"alg", "RS256"}}; - - auto t = QDateTime::currentSecsSinceEpoch(); - QJsonObject payload = {{"identity", getDongleId().value_or("")}, {"nbf", t}, {"iat", t}, {"exp", t + expiry}}; - for (auto it = payloads.begin(); it != payloads.end(); ++it) { - payload.insert(it.key(), it.value()); - } - - auto b64_opts = QByteArray::Base64UrlEncoding | QByteArray::OmitTrailingEquals; - QString jwt = QJsonDocument(header).toJson(QJsonDocument::Compact).toBase64(b64_opts) + '.' + - QJsonDocument(payload).toJson(QJsonDocument::Compact).toBase64(b64_opts); - - auto hash = QCryptographicHash::hash(jwt.toUtf8(), QCryptographicHash::Sha256); - return jwt + "." + rsa_sign(hash).toBase64(b64_opts); -} - -} // namespace CommaApi - -HttpRequest::HttpRequest(QObject *parent, bool create_jwt, int timeout) : create_jwt(create_jwt), QObject(parent) { - networkTimer = new QTimer(this); - networkTimer->setSingleShot(true); - networkTimer->setInterval(timeout); - connect(networkTimer, &QTimer::timeout, this, &HttpRequest::requestTimeout); -} - -bool HttpRequest::active() const { - return reply != nullptr; -} - -bool HttpRequest::timeout() const { - return reply && reply->error() == QNetworkReply::OperationCanceledError; -} - -void HttpRequest::sendRequest(const QString &requestURL, const HttpRequest::Method method) { - if (active()) { - qDebug() << "HttpRequest is active"; - return; - } - QString token; - if (create_jwt) { - token = CommaApi::create_jwt(); - } else { - QString token_json = QString::fromStdString(util::read_file(util::getenv("HOME") + "/.comma/auth.json")); - QJsonDocument json_d = QJsonDocument::fromJson(token_json.toUtf8()); - token = json_d["access_token"].toString(); - } - - QNetworkRequest request; - request.setUrl(QUrl(requestURL)); - request.setRawHeader("User-Agent", getUserAgent().toUtf8()); - - if (!token.isEmpty()) { - request.setRawHeader(QByteArray("Authorization"), ("JWT " + token).toUtf8()); - } - - if (method == HttpRequest::Method::GET) { - reply = nam()->get(request); - } else if (method == HttpRequest::Method::DELETE) { - reply = nam()->deleteResource(request); - } - - networkTimer->start(); - connect(reply, &QNetworkReply::finished, this, &HttpRequest::requestFinished); -} - -void HttpRequest::requestTimeout() { - reply->abort(); -} - -void HttpRequest::requestFinished() { - networkTimer->stop(); - - if (reply->error() == QNetworkReply::NoError) { - emit requestDone(reply->readAll(), true, reply->error()); - } else { - QString error; - if (reply->error() == QNetworkReply::OperationCanceledError) { - nam()->clearAccessCache(); - nam()->clearConnectionCache(); - error = "Request timed out"; - } else { - error = reply->errorString(); - } - emit requestDone(error, false, reply->error()); - } - - reply->deleteLater(); - reply = nullptr; -} - -QNetworkAccessManager *HttpRequest::nam() { - static QNetworkAccessManager *networkAccessManager = new QNetworkAccessManager(qApp); - return networkAccessManager; -} diff --git a/tools/cabana/utils/api.h b/tools/cabana/utils/api.h deleted file mode 100644 index ad64d7e72..000000000 --- a/tools/cabana/utils/api.h +++ /dev/null @@ -1,47 +0,0 @@ -#pragma once - -#include -#include -#include -#include - -#include "common/util.h" - -namespace CommaApi { - -const QString BASE_URL = util::getenv("API_HOST", "https://api.commadotai.com").c_str(); -QByteArray rsa_sign(const QByteArray &data); -QString create_jwt(const QJsonObject &payloads = {}, int expiry = 3600); - -} // namespace CommaApi - -/** - * Makes a request to the request endpoint. - */ - -class HttpRequest : public QObject { - Q_OBJECT - -public: - enum class Method {GET, DELETE}; - - explicit HttpRequest(QObject* parent, bool create_jwt = true, int timeout = 20000); - void sendRequest(const QString &requestURL, const Method method = Method::GET); - bool active() const; - bool timeout() const; - -signals: - void requestDone(const QString &response, bool success, QNetworkReply::NetworkError error); - -protected: - QNetworkReply *reply = nullptr; - -private: - static QNetworkAccessManager *nam(); - QTimer *networkTimer = nullptr; - bool create_jwt; - -private slots: - void requestTimeout(); - void requestFinished(); -}; diff --git a/tools/lib/file_downloader.py b/tools/lib/file_downloader.py new file mode 100755 index 000000000..c9c26bb30 --- /dev/null +++ b/tools/lib/file_downloader.py @@ -0,0 +1,148 @@ +#!/usr/bin/env python3 +""" +CLI tool for downloading files and querying the comma API. +Called by C++ replay/cabana via subprocess. + +Subcommands: + route-files - Get route file URLs as JSON + download - Download URL to local cache, print local path + devices - List user's devices as JSON + device-routes - List routes for a device as JSON +""" +import argparse +import hashlib +import json +import os +import sys +import tempfile +import shutil + +from openpilot.system.hardware.hw import Paths +from openpilot.tools.lib.api import CommaApi, UnauthorizedError, APIError +from openpilot.tools.lib.auth_config import get_token +from openpilot.tools.lib.url_file import URLFile + + +def api_call(func): + """Run an API call, outputting JSON result or error to stdout.""" + try: + result = func(CommaApi(get_token())) + json.dump(result, sys.stdout) + except UnauthorizedError: + json.dump({"error": "unauthorized"}, sys.stdout) + except APIError as e: + error = "not_found" if getattr(e, 'status_code', 0) == 404 else str(e) + json.dump({"error": error}, sys.stdout) + except Exception as e: + json.dump({"error": str(e)}, sys.stdout) + sys.stdout.write("\n") + sys.stdout.flush() + + +def cache_file_path(url): + url_without_query = url.split("?")[0] + return os.path.join(Paths.download_cache_root(), hashlib.sha256(url_without_query.encode()).hexdigest()) + + +def cmd_route_files(args): + api_call(lambda api: api.get(f"v1/route/{args.route}/files")) + + +def cmd_download(args): + url = args.url + use_cache = not args.no_cache + + if use_cache: + local_path = cache_file_path(url) + if os.path.exists(local_path): + sys.stdout.write(local_path + "\n") + sys.stdout.flush() + return + + try: + uf = URLFile(url, cache=False) + total = uf.get_length() + if total <= 0: + sys.stderr.write("ERROR:File not found or empty\n") + sys.stderr.flush() + sys.exit(1) + + os.makedirs(Paths.download_cache_root(), exist_ok=True) + tmp_fd, tmp_path = tempfile.mkstemp(dir=Paths.download_cache_root()) + try: + downloaded = 0 + chunk_size = 1024 * 1024 + with os.fdopen(tmp_fd, 'wb') as f: + while downloaded < total: + data = uf.read(min(chunk_size, total - downloaded)) + f.write(data) + downloaded += len(data) + sys.stderr.write(f"PROGRESS:{downloaded}:{total}\n") + sys.stderr.flush() + + if use_cache: + shutil.move(tmp_path, local_path) + sys.stdout.write(local_path + "\n") + else: + sys.stdout.write(tmp_path + "\n") + except Exception: + try: + os.unlink(tmp_path) + except OSError: + pass + raise + + except Exception as e: + sys.stderr.write(f"ERROR:{e}\n") + sys.stderr.flush() + sys.exit(1) + + sys.stdout.flush() + + +def cmd_devices(args): + api_call(lambda api: api.get("v1/me/devices/")) + + +def cmd_device_routes(args): + def fetch(api): + if args.preserved: + return api.get(f"v1/devices/{args.dongle_id}/routes/preserved") + params = {} + if args.start is not None: + params['start'] = args.start + if args.end is not None: + params['end'] = args.end + return api.get(f"v1/devices/{args.dongle_id}/routes_segments", params=params) + api_call(fetch) + + +def main(): + parser = argparse.ArgumentParser(description="File downloader CLI for openpilot tools") + subparsers = parser.add_subparsers(dest="command", required=True) + + p_rf = subparsers.add_parser("route-files") + p_rf.add_argument("route") + p_rf.set_defaults(func=cmd_route_files) + + p_dl = subparsers.add_parser("download") + p_dl.add_argument("url") + p_dl.add_argument("--no-cache", action="store_true") + p_dl.set_defaults(func=cmd_download) + + p_dev = subparsers.add_parser("devices") + p_dev.set_defaults(func=cmd_devices) + + p_dr = subparsers.add_parser("device-routes") + p_dr.add_argument("dongle_id") + p_dr.add_argument("--start", type=int, default=None) + p_dr.add_argument("--end", type=int, default=None) + p_dr.add_argument("--preserved", action="store_true") + p_dr.set_defaults(func=cmd_device_routes) + + args = parser.parse_args() + args.func(args) + + +if __name__ == "__main__": + main() diff --git a/tools/replay/SConscript b/tools/replay/SConscript index b39cf6dab..3efa970b3 100644 --- a/tools/replay/SConscript +++ b/tools/replay/SConscript @@ -7,12 +7,12 @@ base_frameworks = [] base_libs = [common, messaging, cereal, visionipc, 'm', 'ssl', 'crypto', 'pthread'] replay_lib_src = ["replay.cc", "consoleui.cc", "camera.cc", "filereader.cc", "logreader.cc", "framereader.cc", - "route.cc", "util.cc", "seg_mgr.cc", "timeline.cc", "api.cc"] + "route.cc", "util.cc", "seg_mgr.cc", "timeline.cc", "py_downloader.cc"] if arch != "Darwin": replay_lib_src.append("qcom_decoder.cc") replay_lib = replay_env.Library("replay", replay_lib_src, LIBS=base_libs, FRAMEWORKS=base_frameworks) Export('replay_lib') -replay_libs = [replay_lib, 'avformat', 'avcodec', 'swresample', 'avutil', 'x264', 'z', 'bz2', 'zstd', 'curl', 'yuv', 'ncurses'] + base_libs +replay_libs = [replay_lib, 'avformat', 'avcodec', 'swresample', 'avutil', 'x264', 'z', 'bz2', 'zstd', 'yuv', 'ncurses'] + base_libs replay_env.Program("replay", ["main.cc"], LIBS=replay_libs, FRAMEWORKS=base_frameworks) if GetOption('extras'): diff --git a/tools/replay/api.cc b/tools/replay/api.cc deleted file mode 100644 index 85e4e52b2..000000000 --- a/tools/replay/api.cc +++ /dev/null @@ -1,162 +0,0 @@ - -#include "tools/replay/api.h" - -#include -#include -#include -#include - -#include -#include -#include - -#include "common/params.h" -#include "common/version.h" -#include "system/hardware/hw.h" - -namespace CommaApi2 { - -// Base64 URL-safe character set (uses '-' and '_' instead of '+' and '/') -static const std::string base64url_chars = - "ABCDEFGHIJKLMNOPQRSTUVWXYZ" - "abcdefghijklmnopqrstuvwxyz" - "0123456789-_"; - -std::string base64url_encode(const std::string &in) { - std::string out; - int val = 0, valb = -6; - for (unsigned char c : in) { - val = (val << 8) + c; - valb += 8; - while (valb >= 0) { - out.push_back(base64url_chars[(val >> valb) & 0x3F]); - valb -= 6; - } - } - if (valb > -6) { - out.push_back(base64url_chars[((val << 8) >> (valb + 8)) & 0x3F]); - } - - return out; -} - -EVP_PKEY *get_rsa_private_key() { - static std::unique_ptr rsa_private(nullptr, EVP_PKEY_free); - if (!rsa_private) { - FILE *fp = fopen(Path::rsa_file().c_str(), "rb"); - if (!fp) { - std::cerr << "No RSA private key found, please run manager.py or registration.py" << std::endl; - return nullptr; - } - rsa_private.reset(PEM_read_PrivateKey(fp, NULL, NULL, NULL)); - fclose(fp); - } - return rsa_private.get(); -} - -std::string rsa_sign(const std::string &data) { - EVP_PKEY *private_key = get_rsa_private_key(); - if (!private_key) return {}; - - EVP_MD_CTX *mdctx = EVP_MD_CTX_new(); - assert(mdctx != nullptr); - - std::vector sig(EVP_PKEY_size(private_key)); - uint32_t sig_len; - - EVP_SignInit(mdctx, EVP_sha256()); - EVP_SignUpdate(mdctx, data.data(), data.size()); - int ret = EVP_SignFinal(mdctx, sig.data(), &sig_len, private_key); - - EVP_MD_CTX_free(mdctx); - - assert(ret == 1); - assert(sig.size() == sig_len); - return std::string(sig.begin(), sig.begin() + sig_len); -} - -std::string create_jwt(const json11::Json &extra, int exp_time) { - int now = std::chrono::seconds(std::time(nullptr)).count(); - std::string dongle_id = Params().get("DongleId"); - - // Create header and payload - json11::Json header = json11::Json::object{{"alg", "RS256"}}; - auto payload = json11::Json::object{ - {"identity", dongle_id}, - {"iat", now}, - {"nbf", now}, - {"exp", now + exp_time}, - }; - // Merge extra payload - for (const auto &item : extra.object_items()) { - payload[item.first] = item.second; - } - - // JWT construction - std::string jwt = base64url_encode(header.dump()) + '.' + - base64url_encode(json11::Json(payload).dump()); - - // Hash and sign - std::string hash(SHA256_DIGEST_LENGTH, '\0'); - SHA256((uint8_t *)jwt.data(), jwt.size(), (uint8_t *)hash.data()); - std::string signature = rsa_sign(hash); - - return jwt + "." + base64url_encode(signature); -} - -std::string create_token(bool use_jwt, const json11::Json &payloads, int expiry) { - if (use_jwt) { - return create_jwt(payloads, expiry); - } - - std::string token_json = util::read_file(util::getenv("HOME") + "/.comma/auth.json"); - std::string err; - auto json = json11::Json::parse(token_json, err); - if (!err.empty()) { - std::cerr << "Error parsing auth.json " << err << std::endl; - return ""; - } - return json["access_token"].string_value(); -} - -std::string httpGet(const std::string &url, long *response_code) { - CURL *curl = curl_easy_init(); - assert(curl); - - std::string readBuffer; - const std::string token = CommaApi2::create_token(!Hardware::PC()); - - // Set up the lambda for the write callback - // The '+' makes the lambda non-capturing, allowing it to be used as a C function pointer - auto writeCallback = +[](char *contents, size_t size, size_t nmemb, std::string *userp) ->size_t{ - size_t totalSize = size * nmemb; - userp->append((char *)contents, totalSize); - return totalSize; - }; - - curl_easy_setopt(curl, CURLOPT_URL, url.c_str()); - curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, writeCallback); - curl_easy_setopt(curl, CURLOPT_WRITEDATA, &readBuffer); - curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L); - - // Handle headers - struct curl_slist *headers = nullptr; - headers = curl_slist_append(headers, "User-Agent: openpilot-" COMMA_VERSION); - if (!token.empty()) { - headers = curl_slist_append(headers, ("Authorization: JWT " + token).c_str()); - } - curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers); - - CURLcode res = curl_easy_perform(curl); - - if (response_code) { - curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, response_code); - } - - curl_slist_free_all(headers); - curl_easy_cleanup(curl); - - return res == CURLE_OK ? readBuffer : std::string{}; -} - -} // namespace CommaApi diff --git a/tools/replay/api.h b/tools/replay/api.h deleted file mode 100644 index dff59c065..000000000 --- a/tools/replay/api.h +++ /dev/null @@ -1,15 +0,0 @@ -#pragma once - -#include -#include - -#include "common/util.h" -#include "third_party/json11/json11.hpp" - -namespace CommaApi2 { - -const std::string BASE_URL = util::getenv("API_HOST", "https://api.commadotai.com").c_str(); -std::string create_token(bool use_jwt, const json11::Json& payloads = {}, int expiry = 3600); -std::string httpGet(const std::string &url, long *response_code = nullptr); - -} // namespace CommaApi2 diff --git a/tools/replay/consoleui.cc b/tools/replay/consoleui.cc index 4d43df2da..2d21b4efc 100644 --- a/tools/replay/consoleui.cc +++ b/tools/replay/consoleui.cc @@ -9,6 +9,7 @@ #include "common/ratekeeper.h" #include "common/util.h" #include "common/version.h" +#include "tools/replay/py_downloader.h" namespace { diff --git a/tools/replay/filereader.cc b/tools/replay/filereader.cc index cb13775a3..93a6a1193 100644 --- a/tools/replay/filereader.cc +++ b/tools/replay/filereader.cc @@ -1,49 +1,14 @@ #include "tools/replay/filereader.h" -#include - #include "common/util.h" -#include "system/hardware/hw.h" -#include "tools/replay/util.h" - -std::string cacheFilePath(const std::string &url) { - static std::string cache_path = [] { - const std::string comma_cache = Path::download_cache_root(); - util::create_directories(comma_cache, 0755); - return comma_cache.back() == '/' ? comma_cache : comma_cache + "/"; - }(); - - return cache_path + sha256(getUrlWithoutQuery(url)); -} +#include "tools/replay/py_downloader.h" std::string FileReader::read(const std::string &file, std::atomic *abort) { const bool is_remote = (file.find("https://") == 0) || (file.find("http://") == 0); - const std::string local_file = is_remote ? cacheFilePath(file) : file; - std::string result; - - if ((!is_remote || cache_to_local_) && util::file_exists(local_file)) { - result = util::read_file(local_file); - } else if (is_remote) { - result = download(file, abort); - if (cache_to_local_ && !result.empty()) { - std::ofstream fs(local_file, std::ios::binary | std::ios::out); - fs.write(result.data(), result.size()); - } + if (is_remote) { + std::string local_path = PyDownloader::download(file, cache_to_local_, abort); + if (local_path.empty()) return {}; + return util::read_file(local_path); } - return result; -} - -std::string FileReader::download(const std::string &url, std::atomic *abort) { - for (int i = 0; i <= max_retries_ && !(abort && *abort); ++i) { - if (i > 0) { - rWarning("download failed, retrying %d", i); - util::sleep_for(3000); - } - - std::string result = httpGet(url, chunk_size_, abort); - if (!result.empty()) { - return result; - } - } - return {}; + return util::read_file(file); } diff --git a/tools/replay/filereader.h b/tools/replay/filereader.h index 34aa91e85..30740cd0a 100644 --- a/tools/replay/filereader.h +++ b/tools/replay/filereader.h @@ -5,16 +5,10 @@ class FileReader { public: - FileReader(bool cache_to_local, size_t chunk_size = 0, int retries = 3) - : cache_to_local_(cache_to_local), chunk_size_(chunk_size), max_retries_(retries) {} + FileReader(bool cache_to_local) : cache_to_local_(cache_to_local) {} virtual ~FileReader() {} std::string read(const std::string &file, std::atomic *abort = nullptr); private: - std::string download(const std::string &url, std::atomic *abort); - size_t chunk_size_; - int max_retries_; bool cache_to_local_; }; - -std::string cacheFilePath(const std::string &url); diff --git a/tools/replay/framereader.cc b/tools/replay/framereader.cc index 96ec5915b..13d4497a5 100644 --- a/tools/replay/framereader.cc +++ b/tools/replay/framereader.cc @@ -7,6 +7,7 @@ #include "common/util.h" #include "third_party/libyuv/include/libyuv.h" +#include "tools/replay/py_downloader.h" #include "tools/replay/util.h" #include "system/hardware/hw.h" @@ -71,13 +72,13 @@ FrameReader::~FrameReader() { if (input_ctx) avformat_close_input(&input_ctx); } -bool FrameReader::load(CameraType type, const std::string &url, bool no_hw_decoder, std::atomic *abort, bool local_cache, int chunk_size, int retries) { - auto local_file_path = (url.find("https://") == 0 || url.find("http://") == 0) ? cacheFilePath(url) : url; - if (!util::file_exists(local_file_path)) { - FileReader f(local_cache, chunk_size, retries); - if (f.read(url, abort).empty()) { - return false; - } +bool FrameReader::load(CameraType type, const std::string &url, bool no_hw_decoder, std::atomic *abort, bool local_cache) { + std::string local_file_path; + if (url.find("https://") == 0 || url.find("http://") == 0) { + local_file_path = PyDownloader::download(url, local_cache, abort); + if (local_file_path.empty()) return false; + } else { + local_file_path = url; } return loadFromFile(type, local_file_path, no_hw_decoder, abort); } diff --git a/tools/replay/framereader.h b/tools/replay/framereader.h index d8e86fce0..3609d64f8 100644 --- a/tools/replay/framereader.h +++ b/tools/replay/framereader.h @@ -4,7 +4,6 @@ #include #include "msgq/visionipc/visionbuf.h" -#include "tools/replay/filereader.h" #include "tools/replay/util.h" #ifndef __APPLE__ @@ -22,8 +21,7 @@ class FrameReader { public: FrameReader(); ~FrameReader(); - bool load(CameraType type, const std::string &url, bool no_hw_decoder = false, std::atomic *abort = nullptr, bool local_cache = false, - int chunk_size = -1, int retries = 0); + bool load(CameraType type, const std::string &url, bool no_hw_decoder = false, std::atomic *abort = nullptr, bool local_cache = false); bool loadFromFile(CameraType type, const std::string &file, bool no_hw_decoder = false, std::atomic *abort = nullptr); bool get(int idx, VisionBuf *buf); size_t getFrameCount() const { return packets_info.size(); } diff --git a/tools/replay/logreader.cc b/tools/replay/logreader.cc index 75abb8417..997a4bc00 100644 --- a/tools/replay/logreader.cc +++ b/tools/replay/logreader.cc @@ -6,8 +6,8 @@ #include "tools/replay/util.h" #include "common/util.h" -bool LogReader::load(const std::string &url, std::atomic *abort, bool local_cache, int chunk_size, int retries) { - std::string data = FileReader(local_cache, chunk_size, retries).read(url, abort); +bool LogReader::load(const std::string &url, std::atomic *abort, bool local_cache) { + std::string data = FileReader(local_cache).read(url, abort); if (!data.empty()) { if (url.find(".bz2") != std::string::npos || util::starts_with(data, "BZh9")) { data = decompressBZ2(data, abort); diff --git a/tools/replay/logreader.h b/tools/replay/logreader.h index f8d60ffad..9219878ac 100644 --- a/tools/replay/logreader.h +++ b/tools/replay/logreader.h @@ -28,7 +28,7 @@ class LogReader { public: LogReader(const std::vector &filters = {}) { filters_ = filters; } bool load(const std::string &url, std::atomic *abort = nullptr, - bool local_cache = false, int chunk_size = -1, int retries = 0); + bool local_cache = false); bool load(const char *data, size_t size, std::atomic *abort = nullptr); std::vector events; diff --git a/tools/replay/main.cc b/tools/replay/main.cc index 30f85b170..bdb9cf4f3 100644 --- a/tools/replay/main.cc +++ b/tools/replay/main.cc @@ -1,5 +1,6 @@ #include +#include #include #include #include @@ -132,6 +133,10 @@ int main(int argc, char *argv[]) { util::set_file_descriptor_limit(1024); #endif + // The vendored ncurses static library has a wrong compiled-in terminfo path. + // Point it at the system terminfo database if not already set. + setenv("TERMINFO_DIRS", "/usr/share/terminfo:/lib/terminfo:/usr/lib/terminfo", 0); + ReplayConfig config; if (!parseArgs(argc, argv, config)) { diff --git a/tools/replay/py_downloader.cc b/tools/replay/py_downloader.cc new file mode 100644 index 000000000..efaf3c93a --- /dev/null +++ b/tools/replay/py_downloader.cc @@ -0,0 +1,218 @@ +#include "tools/replay/py_downloader.h" + +#include +#include +#include +#include +#include + +#include "tools/replay/util.h" + +namespace { + +static std::mutex handler_mutex; +static DownloadProgressHandler progress_handler = nullptr; + +// Run a Python command and capture stdout. Optionally parse stderr for PROGRESS lines. +// Returns stdout content. If abort is signaled, kills the child process. +std::string runPython(const std::vector &args, std::atomic *abort = nullptr, bool parse_progress = false) { + // Build argv for execvp + std::vector argv; + argv.push_back("python3"); + argv.push_back("-m"); + argv.push_back("openpilot.tools.lib.file_downloader"); + for (const auto &a : args) { + argv.push_back(a.c_str()); + } + argv.push_back(nullptr); + + int stdout_pipe[2]; + int stderr_pipe[2]; + if (pipe(stdout_pipe) != 0) { + rWarning("py_downloader: pipe() failed"); + return {}; + } + if (pipe(stderr_pipe) != 0) { + rWarning("py_downloader: pipe() failed"); + close(stdout_pipe[0]); close(stdout_pipe[1]); + return {}; + } + + pid_t pid = fork(); + if (pid < 0) { + rWarning("py_downloader: fork() failed"); + close(stdout_pipe[0]); close(stdout_pipe[1]); + close(stderr_pipe[0]); close(stderr_pipe[1]); + return {}; + } + + if (pid == 0) { + // Child process — detach from controlling terminal so Python + // cannot corrupt terminal settings needed by ncurses in the parent. + setsid(); + int devnull = open("/dev/null", O_RDONLY); + if (devnull >= 0) { + dup2(devnull, STDIN_FILENO); + if (devnull > STDERR_FILENO) close(devnull); + } + + // Clear OPENPILOT_PREFIX so the Python process uses default paths + // (e.g. ~/.comma/auth.json). The prefix is only for IPC in the parent. + unsetenv("OPENPILOT_PREFIX"); + + close(stdout_pipe[0]); + close(stderr_pipe[0]); + dup2(stdout_pipe[1], STDOUT_FILENO); + dup2(stderr_pipe[1], STDERR_FILENO); + close(stdout_pipe[1]); + close(stderr_pipe[1]); + + execvp("python3", const_cast(argv.data())); + _exit(127); + } + + // Parent process + close(stdout_pipe[1]); + close(stderr_pipe[1]); + + std::string stdout_data; + std::string stderr_buf; + char buf[4096]; + + // Use select() to read from both pipes + fd_set rfds; + int max_fd = std::max(stdout_pipe[0], stderr_pipe[0]); + bool stdout_open = true, stderr_open = true; + + while (stdout_open || stderr_open) { + if (abort && *abort) { + kill(pid, SIGTERM); + break; + } + + FD_ZERO(&rfds); + if (stdout_open) FD_SET(stdout_pipe[0], &rfds); + if (stderr_open) FD_SET(stderr_pipe[0], &rfds); + + struct timeval tv = {0, 100000}; // 100ms timeout + int ret = select(max_fd + 1, &rfds, nullptr, nullptr, &tv); + if (ret < 0) break; + + if (stdout_open && FD_ISSET(stdout_pipe[0], &rfds)) { + ssize_t n = read(stdout_pipe[0], buf, sizeof(buf)); + if (n <= 0) { + stdout_open = false; + } else { + stdout_data.append(buf, n); + } + } + + if (stderr_open && FD_ISSET(stderr_pipe[0], &rfds)) { + ssize_t n = read(stderr_pipe[0], buf, sizeof(buf)); + if (n <= 0) { + stderr_open = false; + } else { + stderr_buf.append(buf, n); + // Parse complete lines from stderr + size_t pos; + while ((pos = stderr_buf.find('\n')) != std::string::npos) { + std::string line = stderr_buf.substr(0, pos); + stderr_buf.erase(0, pos + 1); + + if (parse_progress && line.rfind("PROGRESS:", 0) == 0) { + // Parse "PROGRESS::" + auto colon1 = line.find(':', 9); + if (colon1 != std::string::npos) { + try { + uint64_t cur = std::stoull(line.c_str() + 9); + uint64_t total = std::stoull(line.c_str() + colon1 + 1); + std::lock_guard lk(handler_mutex); + if (progress_handler) { + progress_handler(cur, total, true); + } + } catch (...) {} + } + } else if (line.rfind("ERROR:", 0) == 0) { + rWarning("py_downloader: %s", line.c_str() + 6); + } + } + } + } + } + + // Drain remaining pipe data to prevent child from blocking on write + for (int fd : {stdout_pipe[0], stderr_pipe[0]}) { + while (read(fd, buf, sizeof(buf)) > 0) {} + close(fd); + } + + int status; + waitpid(pid, &status, 0); + + bool failed = (abort && *abort) || + (WIFEXITED(status) && WEXITSTATUS(status) != 0) || + WIFSIGNALED(status); + if (failed) { + if (WIFEXITED(status) && WEXITSTATUS(status) != 0) { + rWarning("py_downloader: process exited with code %d", WEXITSTATUS(status)); + } else if (WIFSIGNALED(status)) { + rWarning("py_downloader: process killed by signal %d", WTERMSIG(status)); + } + std::lock_guard lk(handler_mutex); + if (progress_handler) { + progress_handler(0, 0, false); + } + return {}; + } + + // Trim trailing newline + while (!stdout_data.empty() && (stdout_data.back() == '\n' || stdout_data.back() == '\r')) { + stdout_data.pop_back(); + } + + return stdout_data; +} + +} // namespace + +void installDownloadProgressHandler(DownloadProgressHandler handler) { + std::lock_guard lk(handler_mutex); + progress_handler = handler; +} + +namespace PyDownloader { + +std::string download(const std::string &url, bool use_cache, std::atomic *abort) { + std::vector args = {"download", url}; + if (!use_cache) { + args.push_back("--no-cache"); + } + return runPython(args, abort, true); +} + +std::string getRouteFiles(const std::string &route) { + return runPython({"route-files", route}); +} + +std::string getDevices() { + return runPython({"devices"}); +} + +std::string getDeviceRoutes(const std::string &dongle_id, int64_t start_ms, int64_t end_ms, bool preserved) { + std::vector args = {"device-routes", dongle_id}; + if (preserved) { + args.push_back("--preserved"); + } else { + if (start_ms > 0) { + args.push_back("--start"); + args.push_back(std::to_string(start_ms)); + } + if (end_ms > 0) { + args.push_back("--end"); + args.push_back(std::to_string(end_ms)); + } + } + return runPython(args); +} + +} // namespace PyDownloader diff --git a/tools/replay/py_downloader.h b/tools/replay/py_downloader.h new file mode 100644 index 000000000..535189784 --- /dev/null +++ b/tools/replay/py_downloader.h @@ -0,0 +1,24 @@ +#pragma once + +#include +#include +#include + +typedef std::function DownloadProgressHandler; +void installDownloadProgressHandler(DownloadProgressHandler handler); + +namespace PyDownloader { + +// Downloads url to local cache, returns local file path. Reports progress via installDownloadProgressHandler. +std::string download(const std::string &url, bool use_cache = true, std::atomic *abort = nullptr); + +// Returns JSON string of route files (same format as /v1/route/.../files API) +std::string getRouteFiles(const std::string &route); + +// Returns JSON string of user's devices +std::string getDevices(); + +// Returns JSON string of device routes +std::string getDeviceRoutes(const std::string &dongle_id, int64_t start_ms = 0, int64_t end_ms = 0, bool preserved = false); + +} // namespace PyDownloader diff --git a/tools/replay/route.cc b/tools/replay/route.cc index ba0082826..e7b8ed6bb 100644 --- a/tools/replay/route.cc +++ b/tools/replay/route.cc @@ -6,7 +6,7 @@ #include "third_party/json11/json11.hpp" #include "system/hardware/hw.h" -#include "tools/replay/api.h" +#include "tools/replay/py_downloader.h" #include "tools/replay/replay.h" #include "tools/replay/util.h" @@ -103,43 +103,44 @@ bool Route::loadFromAutoSource() { return !segments_.empty(); } -bool Route::loadFromServer(int retries) { - const std::string url = CommaApi2::BASE_URL + "/v1/route/" + route_.str + "/files"; - for (int i = 1; i <= retries; ++i) { - long response_code = 0; - std::string result = CommaApi2::httpGet(url, &response_code); - if (response_code == 200) { - return loadFromJson(result); - } - - if (response_code == 401 || response_code == 403) { - rWarning(">> Unauthorized. Authenticate with tools/lib/auth.py <<"); - err_ = RouteLoadError::Unauthorized; - break; - } - if (response_code == 404) { - rWarning("The specified route could not be found on the server."); - err_ = RouteLoadError::FileNotFound; - break; - } - +bool Route::loadFromServer() { + std::string result = PyDownloader::getRouteFiles(route_.str); + if (result.empty()) { err_ = RouteLoadError::NetworkError; - rWarning("Retrying %d/%d", i, retries); - util::sleep_for(3000); - } - - return false; -} - -bool Route::loadFromJson(const std::string &json) { - const static std::regex rx(R"(\/(\d+)\/)"); - std::string err; - auto jsonData = json11::Json::parse(json, err); - if (!err.empty()) { - rWarning("JSON parsing error: %s", err.c_str()); + rWarning("Failed to fetch route files from server"); return false; } - for (const auto &value : jsonData.object_items()) { + + // Check for error field in JSON response + std::string parse_err; + auto json = json11::Json::parse(result, parse_err); + if (!parse_err.empty()) { + err_ = RouteLoadError::NetworkError; + rWarning("Failed to parse route files response"); + return false; + } + + if (json.is_object() && json["error"].is_string()) { + const std::string &error = json["error"].string_value(); + if (error == "unauthorized") { + rWarning(">> Unauthorized. Authenticate with tools/lib/auth.py <<"); + err_ = RouteLoadError::Unauthorized; + } else if (error == "not_found") { + rWarning("The specified route could not be found on the server."); + err_ = RouteLoadError::FileNotFound; + } else { + rWarning("API error: %s", error.c_str()); + err_ = RouteLoadError::NetworkError; + } + return false; + } + + return loadFromJson(json); +} + +bool Route::loadFromJson(const json11::Json &json) { + const static std::regex rx(R"(\/(\d+)\/)"); + for (const auto &value : json.object_items()) { const auto &urlArray = value.second.array_items(); for (const auto &url : urlArray) { std::string url_str = url.string_value(); @@ -225,10 +226,10 @@ void Segment::loadFile(int id, const std::string file) { bool success = false; if (id < MAX_CAMERAS) { frames[id] = std::make_unique(); - success = frames[id]->load((CameraType)id, file, flags & REPLAY_FLAG_NO_HW_DECODER, &abort_, local_cache, 20 * 1024 * 1024, 3); + success = frames[id]->load((CameraType)id, file, flags & REPLAY_FLAG_NO_HW_DECODER, &abort_, local_cache); } else { log = std::make_unique(filters_); - success = log->load(file, &abort_, local_cache, 0, 3); + success = log->load(file, &abort_, local_cache); } if (!success) { diff --git a/tools/replay/route.h b/tools/replay/route.h index 0375252a1..119a81152 100644 --- a/tools/replay/route.h +++ b/tools/replay/route.h @@ -8,6 +8,7 @@ #include #include +#include "third_party/json11/json11.hpp" #include "tools/replay/framereader.h" #include "tools/replay/logreader.h" #include "tools/replay/util.h" @@ -55,8 +56,8 @@ protected: bool loadSegments(); bool loadFromAutoSource(); bool loadFromLocal(); - bool loadFromServer(int retries = 3); - bool loadFromJson(const std::string &json); + bool loadFromServer(); + bool loadFromJson(const json11::Json &json); void addFileToSegment(int seg_num, const std::string &file); RouteIdentifier route_ = {}; std::string data_dir_; diff --git a/tools/replay/tests/test_replay.cc b/tools/replay/tests/test_replay.cc index aed3de59a..45fcc9819 100644 --- a/tools/replay/tests/test_replay.cc +++ b/tools/replay/tests/test_replay.cc @@ -1,5 +1,6 @@ #define CATCH_CONFIG_MAIN #include "catch2/catch.hpp" +#include "tools/replay/filereader.h" #include "tools/replay/replay.h" const std::string TEST_RLOG_URL = "https://commadataci.blob.core.windows.net/openpilotci/0c94aa1e1296d7c6/2021-05-05--19-48-37/0/rlog.bz2"; diff --git a/tools/replay/timeline.cc b/tools/replay/timeline.cc index 39ea8b1be..08dca5c89 100644 --- a/tools/replay/timeline.cc +++ b/tools/replay/timeline.cc @@ -55,7 +55,7 @@ void Timeline::buildTimeline(const Route &route, uint64_t route_start_ts, bool l if (should_exit_) break; auto log = std::make_shared(); - if (!log->load(segment.second.qlog, &should_exit_, local_cache, 0, 3) || log->events.empty()) { + if (!log->load(segment.second.qlog, &should_exit_, local_cache) || log->events.empty()) { continue; // Skip if log loading fails or no events } diff --git a/tools/replay/util.cc b/tools/replay/util.cc index 481564322..7294de828 100644 --- a/tools/replay/util.cc +++ b/tools/replay/util.cc @@ -1,20 +1,13 @@ #include "tools/replay/util.h" #include -#include #include #include -#include -#include #include #include -#include #include -#include #include -#include -#include #include #include "common/timing.h" @@ -51,91 +44,6 @@ void logMessage(ReplyMsgType type, const char *fmt, ...) { free(msg_buf); } -namespace { - -struct CURLGlobalInitializer { - CURLGlobalInitializer() { curl_global_init(CURL_GLOBAL_DEFAULT); } - ~CURLGlobalInitializer() { curl_global_cleanup(); } -}; - -static CURLGlobalInitializer curl_initializer; - -template -struct MultiPartWriter { - T *buf; - size_t *total_written; - size_t offset; - size_t end; - - size_t write(char *data, size_t size, size_t count) { - size_t bytes = size * count; - if ((offset + bytes) > end) return 0; - - if constexpr (std::is_same::value) { - memcpy(buf->data() + offset, data, bytes); - } else if constexpr (std::is_same::value) { - buf->seekp(offset); - buf->write(data, bytes); - } - - offset += bytes; - *total_written += bytes; - return bytes; - } -}; - -template -size_t write_cb(char *data, size_t size, size_t count, void *userp) { - auto w = (MultiPartWriter *)userp; - return w->write(data, size, count); -} - -size_t dumy_write_cb(char *data, size_t size, size_t count, void *userp) { return size * count; } - -struct DownloadStats { - void installDownloadProgressHandler(DownloadProgressHandler handler) { - std::lock_guard lk(lock); - download_progress_handler = handler; - } - - void add(const std::string &url, uint64_t total_bytes) { - std::lock_guard lk(lock); - items[url] = {0, total_bytes}; - } - - void remove(const std::string &url) { - std::lock_guard lk(lock); - items.erase(url); - } - - void update(const std::string &url, uint64_t downloaded, bool success = true) { - std::lock_guard lk(lock); - items[url].first = downloaded; - - auto stat = std::accumulate(items.begin(), items.end(), std::pair{}, [=](auto &a, auto &b){ - return std::pair{a.first + b.second.first, a.second + b.second.second}; - }); - double tm = millis_since_boot(); - if (download_progress_handler && ((tm - prev_tm) > 500 || !success || stat.first >= stat.second)) { - download_progress_handler(stat.first, stat.second, success); - prev_tm = tm; - } - } - - std::mutex lock; - std::map> items; - double prev_tm = 0; - DownloadProgressHandler download_progress_handler = nullptr; -}; - -static DownloadStats download_stats; - -} // namespace - -void installDownloadProgressHandler(DownloadProgressHandler handler) { - download_stats.installDownloadProgressHandler(handler); -} - std::string formattedDataSize(size_t size) { if (size < 1024) { return std::to_string(size) + " B"; @@ -146,140 +54,11 @@ std::string formattedDataSize(size_t size) { } } -size_t getRemoteFileSize(const std::string &url, std::atomic *abort) { - CURL *curl = curl_easy_init(); - if (!curl) return -1; - - curl_easy_setopt(curl, CURLOPT_URL, url.c_str()); - curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, dumy_write_cb); - curl_easy_setopt(curl, CURLOPT_HEADER, 1); - curl_easy_setopt(curl, CURLOPT_NOBODY, 1); - curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1); - curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L); - - CURLM *cm = curl_multi_init(); - curl_multi_add_handle(cm, curl); - int still_running = 1; - while (still_running > 0 && !(abort && *abort)) { - CURLMcode mc = curl_multi_perform(cm, &still_running); - if (mc != CURLM_OK) break; - if (still_running > 0) { - curl_multi_wait(cm, nullptr, 0, 1000, nullptr); - } - } - - double content_length = -1; - curl_easy_getinfo(curl, CURLINFO_CONTENT_LENGTH_DOWNLOAD, &content_length); - curl_multi_remove_handle(cm, curl); - curl_easy_cleanup(curl); - curl_multi_cleanup(cm); - return content_length > 0 ? (size_t)content_length : 0; -} - std::string getUrlWithoutQuery(const std::string &url) { size_t idx = url.find("?"); return (idx == std::string::npos ? url : url.substr(0, idx)); } -template -bool httpDownload(const std::string &url, T &buf, size_t chunk_size, size_t content_length, std::atomic *abort) { - download_stats.add(url, content_length); - - int parts = 1; - if (chunk_size > 0 && content_length > 10 * 1024 * 1024) { - parts = std::nearbyint(content_length / (float)chunk_size); - parts = std::clamp(parts, 1, 5); - } - - CURLM *cm = curl_multi_init(); - size_t written = 0; - std::map> writers; - const int part_size = content_length / parts; - for (int i = 0; i < parts; ++i) { - CURL *eh = curl_easy_init(); - writers[eh] = { - .buf = &buf, - .total_written = &written, - .offset = (size_t)(i * part_size), - .end = i == parts - 1 ? content_length : (i + 1) * part_size, - }; - curl_easy_setopt(eh, CURLOPT_WRITEFUNCTION, write_cb); - curl_easy_setopt(eh, CURLOPT_WRITEDATA, (void *)(&writers[eh])); - curl_easy_setopt(eh, CURLOPT_URL, url.c_str()); - curl_easy_setopt(eh, CURLOPT_RANGE, util::string_format("%d-%d", writers[eh].offset, writers[eh].end - 1).c_str()); - curl_easy_setopt(eh, CURLOPT_HTTPGET, 1); - curl_easy_setopt(eh, CURLOPT_NOSIGNAL, 1); - curl_easy_setopt(eh, CURLOPT_FOLLOWLOCATION, 1); - - curl_multi_add_handle(cm, eh); - } - - int still_running = 1; - size_t prev_written = 0; - while (still_running > 0 && !(abort && *abort)) { - CURLMcode mc = curl_multi_perform(cm, &still_running); - if (mc != CURLM_OK) { - break; - } - if (still_running > 0) { - curl_multi_wait(cm, nullptr, 0, 1000, nullptr); - } - - if (((written - prev_written) / (double)content_length) >= 0.01) { - download_stats.update(url, written); - prev_written = written; - } - } - - CURLMsg *msg; - int msgs_left = -1; - int complete = 0; - while ((msg = curl_multi_info_read(cm, &msgs_left)) && !(abort && *abort)) { - if (msg->msg == CURLMSG_DONE) { - if (msg->data.result == CURLE_OK) { - long res_status = 0; - curl_easy_getinfo(msg->easy_handle, CURLINFO_RESPONSE_CODE, &res_status); - if (res_status == 206) { - complete++; - } else { - rWarning("Download failed: http error code: %d", res_status); - } - } else { - rWarning("Download failed: connection failure: %d", msg->data.result); - } - } - } - - bool success = complete == parts; - download_stats.update(url, written, success); - download_stats.remove(url); - - for (const auto &[e, w] : writers) { - curl_multi_remove_handle(cm, e); - curl_easy_cleanup(e); - } - curl_multi_cleanup(cm); - - return success; -} - -std::string httpGet(const std::string &url, size_t chunk_size, std::atomic *abort) { - size_t size = getRemoteFileSize(url, abort); - if (size == 0) return {}; - - std::string result(size, '\0'); - return httpDownload(url, result, chunk_size, size, abort) ? result : ""; -} - -bool httpDownload(const std::string &url, const std::string &file, size_t chunk_size, std::atomic *abort) { - size_t size = getRemoteFileSize(url, abort); - if (size == 0) return false; - - std::ofstream of(file, std::ios::binary | std::ios::out); - of.seekp(size - 1).write("\0", 1); - return httpDownload(url, of, chunk_size, size, abort); -} - std::string decompressBZ2(const std::string &in, std::atomic *abort) { return decompressBZ2((std::byte *)in.data(), in.size(), abort); } diff --git a/tools/replay/util.h b/tools/replay/util.h index 1f61951d2..ee9219033 100644 --- a/tools/replay/util.h +++ b/tools/replay/util.h @@ -53,12 +53,6 @@ std::string decompressBZ2(const std::byte *in, size_t in_size, std::atomic std::string decompressZST(const std::string &in, std::atomic *abort = nullptr); std::string decompressZST(const std::byte *in, size_t in_size, std::atomic *abort = nullptr); std::string getUrlWithoutQuery(const std::string &url); -size_t getRemoteFileSize(const std::string &url, std::atomic *abort = nullptr); -std::string httpGet(const std::string &url, size_t chunk_size = 0, std::atomic *abort = nullptr); - -typedef std::function DownloadProgressHandler; -void installDownloadProgressHandler(DownloadProgressHandler); -bool httpDownload(const std::string &url, const std::string &file, size_t chunk_size = 0, std::atomic *abort = nullptr); std::string formattedDataSize(size_t size); std::string extractFileName(const std::string& file); std::vector split(std::string_view source, char delimiter);