mirror of
https://github.com/dragonpilot/dragonpilot.git
synced 2026-07-14 01:12:06 +08:00
Vehicle Researcher
6adb63b915
date: 2026-06-04T09:49:56 master commit: c0ab3550eca2e9daf197c46b7e4b24aa9637cf2e
1900 lines
70 KiB
C++
1900 lines
70 KiB
C++
#include "tools/jotpluggler/app.h"
|
|
#include "tools/jotpluggler/car_fingerprint_to_dbc.h"
|
|
#include "tools/jotpluggler/common.h"
|
|
|
|
#include <kj/exception.h>
|
|
|
|
#include <chrono>
|
|
#include <cmath>
|
|
#include <cstdlib>
|
|
#include <cstring>
|
|
#include <iostream>
|
|
#include <initializer_list>
|
|
#include <map>
|
|
#include <mutex>
|
|
#include <limits>
|
|
#include <numeric>
|
|
#include <regex>
|
|
#include <set>
|
|
#include <stdexcept>
|
|
#include <thread>
|
|
#include <tuple>
|
|
#include <unordered_set>
|
|
#include <utility>
|
|
|
|
#include "common/util.h"
|
|
#include "json11/json11.hpp"
|
|
#include "tools/replay/logreader.h"
|
|
#include "tools/replay/py_downloader.h"
|
|
|
|
namespace fs = std::filesystem;
|
|
|
|
namespace {
|
|
|
|
struct RouteSelection {
|
|
std::string dongle_id;
|
|
std::string timestamp;
|
|
int begin_segment = 0;
|
|
int end_segment = -1;
|
|
bool slice_explicit = false;
|
|
LogSelector selector = LogSelector::Auto;
|
|
bool selector_explicit = false;
|
|
std::string canonical_name;
|
|
};
|
|
|
|
struct SegmentLogs {
|
|
std::string rlog;
|
|
std::string qlog;
|
|
std::string fcamera;
|
|
std::string dcamera;
|
|
std::string ecamera;
|
|
std::string qcamera;
|
|
};
|
|
|
|
struct SchemaIndex {
|
|
size_t fixed_series_count = 0;
|
|
std::vector<std::string> fixed_paths;
|
|
|
|
static const SchemaIndex &instance();
|
|
};
|
|
|
|
constexpr size_t INVALID_DYNAMIC_SLOT = std::numeric_limits<size_t>::max();
|
|
|
|
struct SeriesAccumulator {
|
|
explicit SeriesAccumulator(size_t fixed_count = 0) : fixed_series(fixed_count) {}
|
|
|
|
std::vector<RouteSeries> fixed_series;
|
|
std::vector<RouteSeries> dynamic_series;
|
|
std::vector<CanMessageData> can_messages;
|
|
std::unordered_map<std::string, size_t> dynamic_slots;
|
|
std::unordered_map<std::string, std::vector<size_t>> list_scalar_slots;
|
|
std::unordered_map<CanMessageId, size_t, CanMessageIdHash> can_message_slots;
|
|
std::unordered_map<std::string, EnumInfo> enum_info;
|
|
};
|
|
|
|
void append_fixed_scalar_point(RouteSeries *series, double tm, double value);
|
|
void append_dynamic_scalar_point(const std::string &path, double tm, double value, SeriesAccumulator *series);
|
|
RouteSeries *ensure_list_scalar_series(const std::string &base_path, size_t index, SeriesAccumulator *series);
|
|
void append_can_frame(CanServiceKind service,
|
|
uint8_t bus,
|
|
uint32_t address,
|
|
uint16_t bus_time,
|
|
capnp::Data::Reader dat,
|
|
double tm,
|
|
SeriesAccumulator *series);
|
|
void decode_can_frame(const dbc::Database *can_dbc,
|
|
const std::string &service_name,
|
|
uint8_t bus,
|
|
uint32_t address,
|
|
const uint8_t *raw,
|
|
size_t data_size,
|
|
double tm,
|
|
SeriesAccumulator *series);
|
|
|
|
#include "tools/jotpluggler/generated_event_extractors.h"
|
|
|
|
struct LoadedRouteArtifacts {
|
|
std::vector<RouteSeries> series;
|
|
std::vector<CanMessageData> can_messages;
|
|
std::vector<LogEntry> logs;
|
|
std::vector<TimelineEntry> timeline;
|
|
std::unordered_map<std::string, EnumInfo> enum_info;
|
|
};
|
|
|
|
struct RouteMetadata {
|
|
std::string car_fingerprint;
|
|
};
|
|
|
|
struct LoadStats {
|
|
using Clock = std::chrono::steady_clock;
|
|
using TimePoint = Clock::time_point;
|
|
|
|
struct SegmentStats {
|
|
int segment_number = -1;
|
|
std::string log_path;
|
|
double download_seconds = 0.0;
|
|
double decompress_seconds = 0.0;
|
|
double parse_seconds = 0.0;
|
|
double extract_seconds = 0.0;
|
|
size_t compressed_bytes = 0;
|
|
size_t decompressed_bytes = 0;
|
|
size_t event_count = 0;
|
|
size_t series_count = 0;
|
|
bool failed = false;
|
|
};
|
|
|
|
explicit LoadStats(const RouteLoadProgressCallback &callback) : progress(callback) {}
|
|
|
|
void publish(RouteLoadStage stage, size_t segment_index, const std::string &segment_name) {
|
|
if (!progress) {
|
|
return;
|
|
}
|
|
RouteLoadProgress update;
|
|
update.stage = stage;
|
|
update.segment_index = segment_index;
|
|
update.segment_count = segment_count;
|
|
update.current = stage == RouteLoadStage::DownloadingSegment
|
|
? segments_downloaded.load()
|
|
: segments_parsed.load();
|
|
update.total = total_segments.load();
|
|
update.segments_downloaded = segments_downloaded.load();
|
|
update.segments_parsed = segments_parsed.load();
|
|
update.total_segments = total_segments.load();
|
|
update.bytes_downloaded = bytes_downloaded.load();
|
|
update.num_workers = num_workers;
|
|
update.segment_name = segment_name;
|
|
std::lock_guard<std::mutex> lock(progress_mutex);
|
|
progress(update);
|
|
}
|
|
|
|
void print_summary(size_t final_series_count) const {
|
|
const auto secs = [](TimePoint a, TimePoint b) { return std::chrono::duration<double>(b - a).count(); };
|
|
const auto mb = [](size_t bytes) { return static_cast<double>(bytes) / (1024.0 * 1024.0); };
|
|
double dl = 0, dc = 0, pa = 0, ex = 0;
|
|
size_t ev = 0, cb = 0, db = 0;
|
|
for (const auto &s : segments) {
|
|
dl += s.download_seconds; dc += s.decompress_seconds;
|
|
pa += s.parse_seconds; ex += s.extract_seconds;
|
|
ev += s.event_count; cb += s.compressed_bytes; db += s.decompressed_bytes;
|
|
}
|
|
std::cerr << std::fixed << std::setprecision(1)
|
|
<< "route loaded in " << secs(load_start, load_end) << "s (" << segment_count << " segments, " << num_workers << " workers)\n"
|
|
<< " resolve: " << secs(load_start, resolve_end) << "s fetch: " << dl << "s (" << mb(cb) << " MB)"
|
|
<< " decompress: " << dc << "s (" << mb(db) << " MB)\n"
|
|
<< " parse: " << pa << "s (" << ev << " events) extract: " << ex << "s merge: " << secs(merge_start, merge_end) << "s"
|
|
<< " series: " << final_series_count << " paths\n";
|
|
for (const auto &s : segments) {
|
|
std::cerr << " seg " << std::setw(2) << s.segment_number << ": "
|
|
<< (s.failed ? "FAILED" : std::to_string(s.download_seconds) + "s + " + std::to_string(s.parse_seconds)
|
|
+ "s (" + std::to_string(s.event_count) + " ev, " + std::to_string(s.series_count) + " series)") << "\n";
|
|
}
|
|
std::cerr.unsetf(std::ios::floatfield);
|
|
}
|
|
|
|
TimePoint load_start;
|
|
TimePoint resolve_end;
|
|
TimePoint merge_start;
|
|
TimePoint merge_end;
|
|
TimePoint load_end;
|
|
size_t segment_count = 0;
|
|
int num_workers = 1;
|
|
std::vector<SegmentStats> segments;
|
|
std::atomic<size_t> segments_downloaded{0};
|
|
std::atomic<size_t> segments_parsed{0};
|
|
std::atomic<size_t> total_segments{0};
|
|
std::atomic<uint64_t> bytes_downloaded{0};
|
|
RouteLoadProgressCallback progress;
|
|
mutable std::mutex progress_mutex;
|
|
};
|
|
|
|
// Skip individual messages that our local Cap'n Proto schema can't project,
|
|
// such as logs recorded by a newer build.
|
|
template <typename Fn>
|
|
void with_parseable_event(kj::ArrayPtr<const capnp::word> data, Fn &&fn) {
|
|
try {
|
|
capnp::FlatArrayMessageReader event_reader(data);
|
|
fn(event_reader.getRoot<cereal::Event>());
|
|
} catch (const kj::Exception &) {
|
|
return;
|
|
}
|
|
}
|
|
|
|
std::string curve_label(std::string_view series_name) {
|
|
return std::string(series_name.empty() ? std::string_view{"plot"} : series_name);
|
|
}
|
|
|
|
bool parse_segment_number(std::string_view value, int *out) {
|
|
if (value.empty()) return false;
|
|
char *end = nullptr;
|
|
const long parsed = std::strtol(std::string(value).c_str(), &end, 10);
|
|
if (end == nullptr || *end != '\0') return false;
|
|
*out = static_cast<int>(parsed);
|
|
return true;
|
|
}
|
|
|
|
bool is_log_selector_char(char c) {
|
|
return c == 'a' || c == 'r' || c == 'q';
|
|
}
|
|
|
|
LogSelector parse_log_selector_char(char c) {
|
|
switch (c) {
|
|
case 'r': return LogSelector::RLog;
|
|
case 'q': return LogSelector::QLog;
|
|
case 'a':
|
|
default: return LogSelector::Auto;
|
|
}
|
|
}
|
|
|
|
const std::string &selected_log_path(const SegmentLogs &segment, LogSelector selector) {
|
|
switch (selector) {
|
|
case LogSelector::RLog:
|
|
return segment.rlog;
|
|
case LogSelector::QLog:
|
|
return segment.qlog;
|
|
case LogSelector::Auto:
|
|
default:
|
|
return !segment.rlog.empty() ? segment.rlog : segment.qlog;
|
|
}
|
|
}
|
|
|
|
RouteSelection parse_route_selection(std::string route_name) {
|
|
RouteSelection route = {};
|
|
route_name = util::strip(route_name);
|
|
if (route_name.size() >= 2 && route_name[route_name.size() - 2] == '/'
|
|
&& is_log_selector_char(static_cast<char>(std::tolower(route_name.back())))) {
|
|
route.selector = parse_log_selector_char(static_cast<char>(std::tolower(route_name.back())));
|
|
route.selector_explicit = true;
|
|
route_name.resize(route_name.size() - 2);
|
|
}
|
|
static const std::regex pattern(R"(^(([a-z0-9]{16})[|_/])?(.{20})((--|/)((-?\d+(:(-?\d+)?)?)|(:-?\d+)))?$)");
|
|
std::smatch match;
|
|
if (!std::regex_match(route_name, match, pattern)) return route;
|
|
|
|
route.dongle_id = match[2].str();
|
|
route.timestamp = match[3].str();
|
|
route.canonical_name = route.dongle_id + "|" + route.timestamp;
|
|
|
|
const std::string separator = match[5].str();
|
|
const std::string range_str = match[6].str();
|
|
if (!range_str.empty()) {
|
|
route.slice_explicit = true;
|
|
if (separator == "/") {
|
|
size_t pos = range_str.find(':');
|
|
int begin_segment = 0;
|
|
if (!parse_segment_number(range_str.substr(0, pos), &begin_segment)) {
|
|
return {};
|
|
}
|
|
route.begin_segment = begin_segment;
|
|
route.end_segment = begin_segment;
|
|
if (pos != std::string::npos) {
|
|
int end_segment = -1;
|
|
const std::string end_str = range_str.substr(pos + 1);
|
|
if (!end_str.empty() && !parse_segment_number(end_str, &end_segment)) {
|
|
return {};
|
|
}
|
|
route.end_segment = end_str.empty() ? -1 : end_segment;
|
|
}
|
|
} else if (separator == "--") {
|
|
int begin_segment = 0;
|
|
if (!parse_segment_number(range_str, &begin_segment)) return {};
|
|
route.begin_segment = begin_segment;
|
|
}
|
|
}
|
|
return route;
|
|
}
|
|
|
|
void add_log_file_to_segments(std::map<int, SegmentLogs> *segments, int segment_number, const std::string &file) {
|
|
std::string name = extractFileName(file);
|
|
const size_t pos = name.find_last_of("--");
|
|
name = pos != std::string::npos ? name.substr(pos + 2) : name;
|
|
SegmentLogs &segment = (*segments)[segment_number];
|
|
if (name == "rlog.bz2" || name == "rlog.zst" || name == "rlog") {
|
|
segment.rlog = file;
|
|
} else if (name == "qlog.bz2" || name == "qlog.zst" || name == "qlog") {
|
|
segment.qlog = file;
|
|
} else if (name == "fcamera.hevc") {
|
|
segment.fcamera = file;
|
|
} else if (name == "dcamera.hevc") {
|
|
segment.dcamera = file;
|
|
} else if (name == "ecamera.hevc") {
|
|
segment.ecamera = file;
|
|
} else if (name == "qcamera.ts") {
|
|
segment.qcamera = file;
|
|
}
|
|
}
|
|
|
|
std::map<int, SegmentLogs> trim_segments(std::map<int, SegmentLogs> segments, const RouteSelection &route) {
|
|
if (route.begin_segment > 0) {
|
|
segments.erase(segments.begin(), segments.lower_bound(route.begin_segment));
|
|
}
|
|
if (route.end_segment >= 0) {
|
|
segments.erase(segments.upper_bound(route.end_segment), segments.end());
|
|
}
|
|
return segments;
|
|
}
|
|
|
|
std::map<int, SegmentLogs> load_segments_from_json(const json11::Json &json) {
|
|
std::map<int, SegmentLogs> segments;
|
|
static const std::regex rx(R"(\/(\d+)\/)");
|
|
for (const auto &value : json.object_items()) {
|
|
for (const auto &url : value.second.array_items()) {
|
|
const std::string url_str = url.string_value();
|
|
std::smatch match;
|
|
if (!std::regex_search(url_str, match, rx)) continue;
|
|
add_log_file_to_segments(&segments, std::stoi(match[1].str()), url_str);
|
|
}
|
|
}
|
|
return segments;
|
|
}
|
|
|
|
std::map<int, SegmentLogs> load_segments_from_server(const RouteSelection &route) {
|
|
const std::string result = PyDownloader::getRouteFiles(route.canonical_name);
|
|
if (result.empty()) throw std::runtime_error("Failed to fetch route files for " + route.canonical_name);
|
|
|
|
std::string parse_error;
|
|
const auto json = json11::Json::parse(result, parse_error);
|
|
if (!parse_error.empty()) throw std::runtime_error("Failed to parse route file list for " + route.canonical_name);
|
|
if (json.is_object() && json["error"].is_string()) {
|
|
throw std::runtime_error("Route API error for " + route.canonical_name + ": " + json["error"].string_value());
|
|
}
|
|
return load_segments_from_json(json);
|
|
}
|
|
|
|
std::map<int, SegmentLogs> load_segments_from_local(const RouteSelection &route, const std::string &data_dir) {
|
|
std::map<int, SegmentLogs> segments;
|
|
const std::string pattern = route.timestamp + "--";
|
|
for (const auto &entry : fs::directory_iterator(data_dir)) {
|
|
if (!entry.is_directory()) continue;
|
|
const std::string dirname = entry.path().filename().string();
|
|
if (dirname.find(pattern) == std::string::npos) continue;
|
|
const size_t marker = dirname.rfind("--");
|
|
if (marker == std::string::npos) continue;
|
|
int segment_number = 0;
|
|
if (!parse_segment_number(dirname.substr(marker + 2), &segment_number)) {
|
|
continue;
|
|
}
|
|
for (const auto &file : fs::directory_iterator(entry.path())) {
|
|
if (file.is_regular_file()) {
|
|
add_log_file_to_segments(&segments, segment_number, file.path().string());
|
|
}
|
|
}
|
|
}
|
|
return segments;
|
|
}
|
|
|
|
RouteIdentifier make_route_identifier(const RouteSelection &route, const std::map<int, SegmentLogs> &segments) {
|
|
RouteIdentifier route_id;
|
|
route_id.dongle_id = route.dongle_id;
|
|
route_id.log_id = route.timestamp;
|
|
route_id.slice_begin = route.begin_segment;
|
|
route_id.slice_end = route.end_segment;
|
|
route_id.slice_explicit = route.slice_explicit;
|
|
route_id.selector = route.selector;
|
|
route_id.selector_explicit = route.selector_explicit;
|
|
if (!segments.empty()) {
|
|
route_id.available_begin = segments.begin()->first;
|
|
route_id.available_end = segments.rbegin()->first;
|
|
}
|
|
return route_id;
|
|
}
|
|
|
|
std::string detect_dbc_for_fingerprint(std::string_view car_fingerprint) {
|
|
return std::string(dbc_for_car_fingerprint(car_fingerprint));
|
|
}
|
|
|
|
std::vector<std::string> available_dbc_names_impl() {
|
|
std::set<std::string> names;
|
|
for (const fs::path &dbc_dir : {
|
|
repo_root() / "opendbc" / "dbc",
|
|
repo_root() / "tools" / "jotpluggler" / "generated_dbcs",
|
|
}) {
|
|
if (fs::exists(dbc_dir) && fs::is_directory(dbc_dir)) {
|
|
for (const auto &entry : fs::directory_iterator(dbc_dir)) {
|
|
if (!entry.is_regular_file() || entry.path().extension() != ".dbc") {
|
|
continue;
|
|
}
|
|
names.insert(entry.path().stem().string());
|
|
}
|
|
}
|
|
}
|
|
for (const auto &[_, dbc_name] : kCarFingerprintToDbc) {
|
|
if (!dbc_name.empty()) {
|
|
names.insert(std::string(dbc_name));
|
|
}
|
|
}
|
|
return std::vector<std::string>(names.begin(), names.end());
|
|
}
|
|
|
|
fs::path resolve_dbc_path(const std::string &dbc_name) {
|
|
for (const fs::path &candidate : {
|
|
repo_root() / "opendbc" / "dbc" / (dbc_name + ".dbc"),
|
|
repo_root() / "tools" / "jotpluggler" / "generated_dbcs" / (dbc_name + ".dbc"),
|
|
}) {
|
|
if (fs::exists(candidate)) return candidate;
|
|
}
|
|
throw std::runtime_error("DBC not found: " + dbc_name);
|
|
}
|
|
|
|
std::array<uint8_t, 3> parse_color(std::string_view color) {
|
|
if (!color.empty() && color.front() == '#') {
|
|
color.remove_prefix(1);
|
|
}
|
|
if (color.size() != 6) return {160, 170, 180};
|
|
|
|
std::array<uint8_t, 3> out = {};
|
|
for (size_t i = 0; i < 3; ++i) {
|
|
const std::string byte(color.substr(i * 2, 2));
|
|
char *end = nullptr;
|
|
const long parsed = std::strtol(byte.c_str(), &end, 16);
|
|
if (end == nullptr || *end != '\0' || parsed < 0 || parsed > 255) return {160, 170, 180};
|
|
out[i] = static_cast<uint8_t>(parsed);
|
|
}
|
|
return out;
|
|
}
|
|
|
|
uint8_t android_priority_to_level(uint8_t priority) {
|
|
switch (priority) {
|
|
case 2:
|
|
case 3:
|
|
return 10;
|
|
case 4:
|
|
return 20;
|
|
case 5:
|
|
return 30;
|
|
case 6:
|
|
return 40;
|
|
case 7:
|
|
default:
|
|
return 50;
|
|
}
|
|
}
|
|
|
|
uint8_t alert_status_to_level(cereal::SelfdriveState::AlertStatus status) {
|
|
switch (status) {
|
|
case cereal::SelfdriveState::AlertStatus::NORMAL:
|
|
return 20;
|
|
case cereal::SelfdriveState::AlertStatus::USER_PROMPT:
|
|
return 30;
|
|
case cereal::SelfdriveState::AlertStatus::CRITICAL:
|
|
return 40;
|
|
}
|
|
return 20;
|
|
}
|
|
|
|
TimelineEntry::Type alert_status_to_timeline_type(cereal::SelfdriveState::AlertStatus status, bool enabled) {
|
|
if (!enabled) {
|
|
return TimelineEntry::Type::None;
|
|
}
|
|
switch (status) {
|
|
case cereal::SelfdriveState::AlertStatus::NORMAL:
|
|
return TimelineEntry::Type::Engaged;
|
|
case cereal::SelfdriveState::AlertStatus::USER_PROMPT:
|
|
return TimelineEntry::Type::AlertInfo;
|
|
case cereal::SelfdriveState::AlertStatus::CRITICAL:
|
|
return TimelineEntry::Type::AlertCritical;
|
|
}
|
|
return TimelineEntry::Type::Engaged;
|
|
}
|
|
|
|
void append_timeline_entry(std::vector<TimelineEntry> *timeline, double mono_time, TimelineEntry::Type type) {
|
|
if (timeline == nullptr) {
|
|
return;
|
|
}
|
|
if (!timeline->empty() && timeline->back().type == type) {
|
|
timeline->back().end_time = std::max(timeline->back().end_time, mono_time);
|
|
return;
|
|
}
|
|
timeline->push_back(TimelineEntry{
|
|
.start_time = mono_time,
|
|
.end_time = mono_time,
|
|
.type = type,
|
|
});
|
|
}
|
|
|
|
double android_wall_time_seconds(uint64_t timestamp) {
|
|
if (timestamp == 0) return 0.0;
|
|
if (timestamp > 1000000000000ULL) return static_cast<double>(timestamp) / 1.0e9;
|
|
if (timestamp > 1000000000ULL) return static_cast<double>(timestamp) / 1.0e6;
|
|
return static_cast<double>(timestamp);
|
|
}
|
|
|
|
std::optional<uint64_t> json_u64_value(const json11::Json &value) {
|
|
if (value.is_number()) {
|
|
const double number = value.number_value();
|
|
if (number >= 0.0) return static_cast<uint64_t>(number);
|
|
}
|
|
if (value.is_string()) {
|
|
try {
|
|
return static_cast<uint64_t>(std::stoull(value.string_value()));
|
|
} catch (...) {
|
|
}
|
|
}
|
|
return std::nullopt;
|
|
}
|
|
|
|
std::optional<int> json_int_value(const json11::Json &value) {
|
|
if (value.is_number()) return value.int_value();
|
|
if (value.is_string()) {
|
|
try {
|
|
return std::stoi(value.string_value());
|
|
} catch (...) {
|
|
}
|
|
}
|
|
return std::nullopt;
|
|
}
|
|
|
|
std::string json_value_for_log(const json11::Json &value) {
|
|
if (value.is_string()) return value.string_value();
|
|
if (value.is_bool()) return value.bool_value() ? "true" : "false";
|
|
return value.dump();
|
|
}
|
|
|
|
std::string format_journal_context(const json11::Json &parsed, int pid, int tid) {
|
|
std::vector<std::string> lines;
|
|
if (pid != 0 || tid != 0) {
|
|
lines.push_back("pid=" + std::to_string(pid) + ", tid=" + std::to_string(tid));
|
|
}
|
|
|
|
const std::array<const char *, 5> preferred_keys = {
|
|
"_HOSTNAME",
|
|
"_TRANSPORT",
|
|
"PRIORITY",
|
|
"SYSLOG_FACILITY",
|
|
"__MONOTONIC_TIMESTAMP",
|
|
};
|
|
for (const char *key : preferred_keys) {
|
|
const json11::Json &value = parsed[key];
|
|
if (!value.is_null()) {
|
|
lines.push_back(std::string(key) + "=" + json_value_for_log(value));
|
|
}
|
|
}
|
|
return join(lines, "\n");
|
|
}
|
|
|
|
std::string alert_message_text(const cereal::SelfdriveState::Reader &state) {
|
|
std::string text = state.getAlertText1().cStr();
|
|
const std::string text2 = state.getAlertText2().cStr();
|
|
if (!text2.empty()) {
|
|
text += " - " + text2;
|
|
}
|
|
return text;
|
|
}
|
|
|
|
bool same_log_entry(const LogEntry &a, const LogEntry &b) {
|
|
return a.mono_time == b.mono_time
|
|
&& a.level == b.level
|
|
&& a.source == b.source
|
|
&& a.func == b.func
|
|
&& a.message == b.message
|
|
&& a.context == b.context
|
|
&& a.origin == b.origin;
|
|
}
|
|
|
|
void append_log_event(cereal::Event::Which which,
|
|
const cereal::Event::Reader &event,
|
|
double time_offset,
|
|
std::vector<LogEntry> *logs,
|
|
std::string *last_alert_key) {
|
|
const double boot_time = static_cast<double>(event.getLogMonoTime()) / 1.0e9;
|
|
const double mono_time = boot_time - time_offset;
|
|
|
|
auto make_entry = [&](LogOrigin origin, uint8_t level = 20) {
|
|
LogEntry e;
|
|
e.mono_time = mono_time;
|
|
e.boot_time = boot_time;
|
|
e.origin = origin;
|
|
e.level = level;
|
|
return e;
|
|
};
|
|
|
|
switch (which) {
|
|
case cereal::Event::Which::LOG_MESSAGE:
|
|
case cereal::Event::Which::ERROR_LOG_MESSAGE: {
|
|
const std::string raw = which == cereal::Event::Which::LOG_MESSAGE
|
|
? event.getLogMessage().cStr() : event.getErrorLogMessage().cStr();
|
|
auto entry = make_entry(LogOrigin::Log, which == cereal::Event::Which::ERROR_LOG_MESSAGE ? 40 : 20);
|
|
entry.source = "log";
|
|
entry.message = raw;
|
|
std::string err;
|
|
if (const auto p = json11::Json::parse(raw, err); err.empty() && p.is_object()) {
|
|
entry.wall_time = p["created"].number_value();
|
|
if (p["levelnum"].is_number()) entry.level = static_cast<uint8_t>(p["levelnum"].int_value());
|
|
const std::string fn = p["filename"].string_value();
|
|
const int ln = p["lineno"].is_number() ? p["lineno"].int_value() : 0;
|
|
entry.source = fn.empty() ? "log" : fn + (ln > 0 ? ":" + std::to_string(ln) : "");
|
|
entry.func = p["funcname"].string_value();
|
|
if (p["msg"].is_string()) entry.message = p["msg"].string_value();
|
|
if (!p["ctx"].is_null()) entry.context = p["ctx"].dump();
|
|
}
|
|
logs->push_back(std::move(entry));
|
|
break;
|
|
}
|
|
case cereal::Event::Which::ANDROID_LOG: {
|
|
const auto android = event.getAndroidLog();
|
|
auto entry = make_entry(LogOrigin::Android, android_priority_to_level(android.getPriority()));
|
|
entry.wall_time = android_wall_time_seconds(android.getTs());
|
|
entry.source = android.hasTag() ? android.getTag().cStr() : "android";
|
|
entry.message = android.hasMessage() ? android.getMessage().cStr() : std::string();
|
|
entry.context = "pid=" + std::to_string(android.getPid()) + ", tid=" + std::to_string(android.getTid());
|
|
if (!entry.message.empty()) {
|
|
std::string err;
|
|
if (const auto p = json11::Json::parse(entry.message, err); err.empty() && p.is_object()) {
|
|
if (p["MESSAGE"].is_string()) entry.message = p["MESSAGE"].string_value();
|
|
if (p["SYSLOG_IDENTIFIER"].is_string() && !p["SYSLOG_IDENTIFIER"].string_value().empty())
|
|
entry.source = p["SYSLOG_IDENTIFIER"].string_value();
|
|
if (auto pri = json_int_value(p["PRIORITY"]); pri.has_value())
|
|
entry.level = android_priority_to_level(*pri);
|
|
if (auto ts = json_u64_value(p["__REALTIME_TIMESTAMP"]); ts.has_value())
|
|
entry.wall_time = android_wall_time_seconds(*ts);
|
|
entry.context = format_journal_context(p, android.getPid(), android.getTid());
|
|
}
|
|
}
|
|
logs->push_back(std::move(entry));
|
|
break;
|
|
}
|
|
case cereal::Event::Which::SELFDRIVE_STATE: {
|
|
const auto sd = event.getSelfdriveState();
|
|
const std::string alert_type = sd.getAlertType().cStr();
|
|
const std::string alert_text1 = sd.getAlertText1().cStr();
|
|
if (alert_text1.empty() && alert_type.empty()) break;
|
|
const std::string key = alert_type + "\n" + alert_text1 + "\n" + std::string(sd.getAlertText2().cStr());
|
|
if (last_alert_key != nullptr && key == *last_alert_key) break;
|
|
if (last_alert_key != nullptr) *last_alert_key = key;
|
|
auto entry = make_entry(LogOrigin::Alert, alert_status_to_level(sd.getAlertStatus()));
|
|
entry.source = "alert";
|
|
entry.func = alert_type;
|
|
entry.message = alert_message_text(sd);
|
|
logs->push_back(std::move(entry));
|
|
break;
|
|
}
|
|
default:
|
|
break;
|
|
}
|
|
}
|
|
|
|
std::vector<TimelineEntry> extract_segment_timeline(const std::vector<Event> &events) {
|
|
std::vector<TimelineEntry> timeline;
|
|
timeline.reserve(events.size() / 16);
|
|
|
|
for (const Event &event_record : events) {
|
|
if (event_record.which != cereal::Event::Which::SELFDRIVE_STATE) {
|
|
continue;
|
|
}
|
|
with_parseable_event(event_record.data, [&](const cereal::Event::Reader &event) {
|
|
const auto sd = event.getSelfdriveState();
|
|
const double mono_time = static_cast<double>(event.getLogMonoTime()) / 1.0e9;
|
|
append_timeline_entry(&timeline, mono_time, alert_status_to_timeline_type(sd.getAlertStatus(), sd.getEnabled()));
|
|
});
|
|
}
|
|
|
|
return timeline;
|
|
}
|
|
|
|
std::vector<LogEntry> extract_segment_logs(const std::vector<Event> &events) {
|
|
std::vector<LogEntry> logs;
|
|
logs.reserve(events.size() / 8);
|
|
std::string last_alert_key;
|
|
|
|
for (const Event &event_record : events) {
|
|
with_parseable_event(event_record.data, [&](const cereal::Event::Reader &event) {
|
|
append_log_event(event_record.which, event, 0.0, &logs, &last_alert_key);
|
|
});
|
|
}
|
|
|
|
return logs;
|
|
}
|
|
|
|
RouteMetadata extract_segment_metadata(const std::vector<Event> &events) {
|
|
RouteMetadata metadata;
|
|
for (const Event &event_record : events) {
|
|
if (event_record.which != cereal::Event::Which::CAR_PARAMS) continue;
|
|
with_parseable_event(event_record.data, [&](const cereal::Event::Reader &event) {
|
|
metadata.car_fingerprint = event.getCarParams().getCarFingerprint().cStr();
|
|
});
|
|
if (!metadata.car_fingerprint.empty()) break;
|
|
}
|
|
return metadata;
|
|
}
|
|
|
|
RouteMetadata detect_route_metadata(const std::map<int, SegmentLogs> &segments, LogSelector selector) {
|
|
for (const auto &[_, segment] : segments) {
|
|
const std::string &log_path = selector == LogSelector::Auto
|
|
? (!segment.qlog.empty() ? segment.qlog : segment.rlog)
|
|
: selected_log_path(segment, selector);
|
|
if (log_path.empty()) {
|
|
continue;
|
|
}
|
|
LogReader reader;
|
|
if (!reader.load(log_path, nullptr, true)) continue;
|
|
RouteMetadata metadata = extract_segment_metadata(reader.events);
|
|
if (!metadata.car_fingerprint.empty()) return metadata;
|
|
}
|
|
return {};
|
|
}
|
|
|
|
std::vector<double> normalize_sizes(const json11::Json &sizes_json, size_t child_count) {
|
|
std::vector<double> parsed;
|
|
if (sizes_json.is_array()) {
|
|
for (const json11::Json &value : sizes_json.array_items()) {
|
|
if (value.is_number()) {
|
|
parsed.push_back(std::max(value.number_value(), 0.0));
|
|
}
|
|
}
|
|
}
|
|
|
|
if (parsed.size() != child_count || child_count == 0) return std::vector<double>(child_count, child_count == 0 ? 0.0 : 1.0 / static_cast<double>(child_count));
|
|
|
|
const double total = std::accumulate(parsed.begin(), parsed.end(), 0.0);
|
|
if (total <= 0.0) return std::vector<double>(child_count, 1.0 / static_cast<double>(child_count));
|
|
for (double &value : parsed) {
|
|
value /= total;
|
|
}
|
|
return parsed;
|
|
}
|
|
|
|
PlotRange parse_range(const json11::Json &pane_node) {
|
|
PlotRange range;
|
|
const json11::Json &range_node = pane_node["range"];
|
|
if (range_node.is_object()) {
|
|
range.valid = true;
|
|
range.left = range_node["left"].number_value();
|
|
range.right = range_node["right"].number_value();
|
|
range.bottom = range_node["bottom"].number_value();
|
|
range.top = range_node["top"].is_number() ? range_node["top"].number_value() : 1.0;
|
|
}
|
|
const json11::Json &limit_y_node = pane_node["y_limits"];
|
|
if (limit_y_node.is_object()) {
|
|
if (limit_y_node["min"].is_number()) {
|
|
range.has_y_limit_min = true;
|
|
range.y_limit_min = limit_y_node["min"].number_value();
|
|
}
|
|
if (limit_y_node["max"].is_number()) {
|
|
range.has_y_limit_max = true;
|
|
range.y_limit_max = limit_y_node["max"].number_value();
|
|
}
|
|
}
|
|
return range;
|
|
}
|
|
|
|
Curve parse_curve(const json11::Json &curve_node) {
|
|
Curve curve;
|
|
curve.name = curve_node["name"].string_value();
|
|
curve.label = curve_label(curve.name);
|
|
curve.color = parse_color(curve_node["color"].string_value());
|
|
|
|
const std::string transform_name = curve_node["transform"].string_value();
|
|
if (transform_name == "derivative") {
|
|
curve.derivative = true;
|
|
curve.derivative_dt = curve_node["derivative_dt"].is_number() ? curve_node["derivative_dt"].number_value() : 0.0;
|
|
} else if (transform_name == "scale") {
|
|
curve.value_scale = curve_node["scale"].is_number() ? curve_node["scale"].number_value() : 1.0;
|
|
curve.value_offset = curve_node["offset"].is_number() ? curve_node["offset"].number_value() : 0.0;
|
|
}
|
|
const json11::Json &custom_node = curve_node["custom_python"];
|
|
if (custom_node.is_object()) {
|
|
CustomPythonSeries spec;
|
|
spec.linked_source = custom_node["linked_source"].string_value();
|
|
spec.globals_code = custom_node["globals_code"].string_value();
|
|
spec.function_code = custom_node["function_code"].string_value();
|
|
for (const json11::Json &source : custom_node["additional_sources"].array_items()) {
|
|
if (source.is_string()) {
|
|
spec.additional_sources.push_back(source.string_value());
|
|
}
|
|
}
|
|
curve.custom_python = std::move(spec);
|
|
}
|
|
return curve;
|
|
}
|
|
|
|
std::string pane_title(const json11::Json &dock_area_node) {
|
|
const std::string raw = dock_area_node["title"].string_value();
|
|
return raw.empty() ? "..." : raw;
|
|
}
|
|
|
|
Pane parse_dock_area(const json11::Json &dock_area_node) {
|
|
Pane pane;
|
|
const std::string kind = dock_area_node["kind"].string_value();
|
|
if (kind == "map") {
|
|
pane.kind = PaneKind::Map;
|
|
} else if (kind == "camera") {
|
|
pane.kind = PaneKind::Camera;
|
|
const std::string camera_view = dock_area_node["camera_view"].string_value();
|
|
if (const CameraViewSpec *spec = camera_view_spec_from_layout_name(camera_view)) {
|
|
pane.camera_view = spec->view;
|
|
} else {
|
|
pane.camera_view = CameraViewKind::Road;
|
|
}
|
|
}
|
|
pane.range = parse_range(dock_area_node);
|
|
const json11::Json &curves_node = dock_area_node["curves"];
|
|
if (curves_node.is_array()) {
|
|
for (const json11::Json &curve_node : curves_node.array_items()) {
|
|
if (curve_node.is_object()) {
|
|
pane.curves.push_back(parse_curve(curve_node));
|
|
}
|
|
}
|
|
}
|
|
pane.title = pane_title(dock_area_node);
|
|
return pane;
|
|
}
|
|
|
|
WorkspaceNode parse_workspace_node(const json11::Json &node, WorkspaceTab *tab) {
|
|
WorkspaceNode workspace_node;
|
|
if (!node.is_object()) return workspace_node;
|
|
|
|
if (node["curves"].is_array()) {
|
|
workspace_node.is_pane = true;
|
|
workspace_node.pane_index = static_cast<int>(tab->panes.size());
|
|
tab->panes.push_back(parse_dock_area(node));
|
|
return workspace_node;
|
|
}
|
|
|
|
const json11::Json &children_node = node["children"];
|
|
if (!children_node.is_array()) return workspace_node;
|
|
|
|
const std::vector<json11::Json> children = children_node.array_items();
|
|
if (children.empty()) return workspace_node;
|
|
|
|
const std::string split = node["split"].string_value();
|
|
workspace_node.orientation = split == "vertical" ? SplitOrientation::Vertical : SplitOrientation::Horizontal;
|
|
const std::vector<double> sizes = normalize_sizes(node["sizes"], children.size());
|
|
workspace_node.sizes.reserve(sizes.size());
|
|
workspace_node.children.reserve(children.size());
|
|
for (size_t i = 0; i < children.size(); ++i) {
|
|
workspace_node.sizes.push_back(static_cast<float>(sizes[i]));
|
|
workspace_node.children.push_back(parse_workspace_node(children[i], tab));
|
|
}
|
|
return workspace_node;
|
|
}
|
|
|
|
WorkspaceTab parse_tab(const json11::Json &tab, const fs::path &layout_path) {
|
|
WorkspaceTab workspace_tab;
|
|
workspace_tab.tab_name = tab["name"].string_value().empty() ? "tab1" : tab["name"].string_value();
|
|
const json11::Json &dock_root = tab["root"];
|
|
if (!dock_root.is_object()) throw std::runtime_error("Layout tab has no dock content: " + layout_path.string());
|
|
workspace_tab.root = parse_workspace_node(dock_root, &workspace_tab);
|
|
return workspace_tab;
|
|
}
|
|
|
|
SketchLayout parse_layout(const fs::path &layout_path) {
|
|
const std::string text = util::read_file(layout_path.string());
|
|
if (text.empty()) throw std::runtime_error("Failed to read layout JSON: " + layout_path.string());
|
|
|
|
std::string parse_error;
|
|
const json11::Json root = json11::Json::parse(text, parse_error);
|
|
if (!parse_error.empty() || !root.is_object()) {
|
|
throw std::runtime_error("Failed to parse layout JSON: " + layout_path.string());
|
|
}
|
|
SketchLayout layout;
|
|
for (const json11::Json &tab : root["tabs"].array_items()) {
|
|
if (tab.is_object()) {
|
|
layout.tabs.push_back(parse_tab(tab, layout_path));
|
|
}
|
|
}
|
|
if (layout.tabs.empty()) throw std::runtime_error("Layout has no tabs: " + layout_path.string());
|
|
const json11::Json &tab_index = root["current_tab_index"].is_number() ? root["current_tab_index"] : root["currentTabIndex"];
|
|
layout.current_tab_index = std::clamp(tab_index.is_number() ? tab_index.int_value() : 0,
|
|
0,
|
|
static_cast<int>(layout.tabs.size()) - 1);
|
|
return layout;
|
|
}
|
|
|
|
const SchemaIndex &SchemaIndex::instance() {
|
|
static const SchemaIndex index = [] {
|
|
SchemaIndex out;
|
|
out.fixed_paths = static_event_fixed_paths();
|
|
out.fixed_series_count = out.fixed_paths.size();
|
|
return out;
|
|
}();
|
|
return index;
|
|
}
|
|
|
|
bool is_absolute_curve(const std::string &name) {
|
|
return !name.empty() && name.front() == '/';
|
|
}
|
|
|
|
void append_scalar_point(RouteSeries *series,
|
|
const std::string &path,
|
|
double tm,
|
|
double value) {
|
|
if (series->path.empty()) {
|
|
series->path = path;
|
|
}
|
|
series->times.push_back(tm);
|
|
series->values.push_back(value);
|
|
}
|
|
|
|
void append_fixed_scalar_point(RouteSeries *series, double tm, double value) {
|
|
series->times.push_back(tm);
|
|
series->values.push_back(value);
|
|
}
|
|
|
|
CanMessageData *ensure_can_message(CanServiceKind service, uint8_t bus, uint32_t address, SeriesAccumulator *series) {
|
|
const CanMessageId id{service, bus, address};
|
|
auto [it, inserted] = series->can_message_slots.try_emplace(id, series->can_messages.size());
|
|
if (inserted) {
|
|
series->can_messages.push_back(CanMessageData{.id = id});
|
|
}
|
|
return &series->can_messages[it->second];
|
|
}
|
|
|
|
void append_can_frame(CanServiceKind service,
|
|
uint8_t bus,
|
|
uint32_t address,
|
|
uint16_t bus_time,
|
|
capnp::Data::Reader dat,
|
|
double tm,
|
|
SeriesAccumulator *series) {
|
|
CanMessageData *message = ensure_can_message(service, bus, address, series);
|
|
message->samples.push_back(CanFrameSample{
|
|
.mono_time = tm,
|
|
.bus_time = bus_time,
|
|
.data = std::string(reinterpret_cast<const char *>(dat.begin()), dat.size()),
|
|
});
|
|
}
|
|
|
|
void append_dynamic_scalar_point(const std::string &path, double tm, double value, SeriesAccumulator *series);
|
|
|
|
void decode_can_frame(const dbc::Database *can_dbc,
|
|
const std::string &service_name,
|
|
uint8_t bus,
|
|
uint32_t address,
|
|
const uint8_t *raw,
|
|
size_t data_size,
|
|
double tm,
|
|
SeriesAccumulator *series) {
|
|
if (can_dbc == nullptr) {
|
|
return;
|
|
}
|
|
const dbc::Message *message = can_dbc->message(address);
|
|
if (message == nullptr) {
|
|
return;
|
|
}
|
|
const std::string base_path = "/" + service_name + "/" + std::to_string(bus) + "/" + message->name;
|
|
for (const dbc::Signal &signal : message->signals) {
|
|
std::optional<double> value = dbc::signalValue(signal, *message, raw, data_size);
|
|
if (!value.has_value()) continue;
|
|
const std::string path = base_path + "/" + signal.name;
|
|
append_dynamic_scalar_point(path, tm, *value, series);
|
|
if (series->enum_info.find(path) == series->enum_info.end()) {
|
|
std::vector<std::string> enum_names = can_dbc->enumNames(signal);
|
|
if (!enum_names.empty()) {
|
|
series->enum_info.emplace(path, EnumInfo{.names = std::move(enum_names)});
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
void append_live_can_frame(CanServiceKind service,
|
|
const LiveCanFrame &frame,
|
|
double time_offset,
|
|
const dbc::Database *can_dbc,
|
|
SeriesAccumulator *series) {
|
|
const double tm = frame.mono_time - time_offset;
|
|
CanMessageData *message = ensure_can_message(service, frame.bus, frame.address, series);
|
|
message->samples.push_back(CanFrameSample{
|
|
.mono_time = tm,
|
|
.bus_time = frame.bus_time,
|
|
.data = frame.data,
|
|
});
|
|
decode_can_frame(can_dbc,
|
|
service == CanServiceKind::Can ? "can" : "sendcan",
|
|
frame.bus,
|
|
frame.address,
|
|
reinterpret_cast<const uint8_t *>(frame.data.data()),
|
|
frame.data.size(),
|
|
tm,
|
|
series);
|
|
}
|
|
|
|
SeriesAccumulator make_series_accumulator(const SchemaIndex &schema) {
|
|
SeriesAccumulator out(schema.fixed_series_count);
|
|
for (size_t i = 0; i < schema.fixed_paths.size(); ++i) {
|
|
out.fixed_series[i].path = schema.fixed_paths[i];
|
|
}
|
|
return out;
|
|
}
|
|
|
|
size_t ensure_dynamic_slot(const std::string &path, SeriesAccumulator *series) {
|
|
auto [it, inserted] = series->dynamic_slots.try_emplace(path, series->dynamic_series.size());
|
|
if (inserted) {
|
|
series->dynamic_series.push_back(RouteSeries{it->first});
|
|
}
|
|
return it->second;
|
|
}
|
|
|
|
RouteSeries *ensure_dynamic_series(const std::string &path, SeriesAccumulator *series) {
|
|
return &series->dynamic_series[ensure_dynamic_slot(path, series)];
|
|
}
|
|
|
|
RouteSeries *ensure_list_scalar_series(const std::string &base_path, size_t index, SeriesAccumulator *series) {
|
|
auto [it, _] = series->list_scalar_slots.try_emplace(base_path);
|
|
std::vector<size_t> &slots = it->second;
|
|
if (slots.size() <= index) {
|
|
slots.resize(index + 1, INVALID_DYNAMIC_SLOT);
|
|
}
|
|
if (slots[index] == INVALID_DYNAMIC_SLOT) {
|
|
slots[index] = ensure_dynamic_slot(base_path + "/" + std::to_string(index), series);
|
|
}
|
|
return &series->dynamic_series[slots[index]];
|
|
}
|
|
|
|
void append_dynamic_scalar_point(const std::string &path, double tm, double value, SeriesAccumulator *series) {
|
|
append_scalar_point(ensure_dynamic_series(path, series), path, tm, value);
|
|
}
|
|
|
|
void append_event_fast(cereal::Event::Which which,
|
|
int32_t eidx_segnum,
|
|
kj::ArrayPtr<const capnp::word> data,
|
|
const dbc::Database *can_dbc,
|
|
bool skip_raw_can,
|
|
double time_offset,
|
|
SeriesAccumulator *series) {
|
|
if (eidx_segnum != -1) {
|
|
return;
|
|
}
|
|
with_parseable_event(data, [&](const cereal::Event::Reader &event) {
|
|
append_event_static_reader(which, event, can_dbc, skip_raw_can, time_offset, series);
|
|
});
|
|
}
|
|
|
|
void append_events_fast_range(const std::vector<Event> &events,
|
|
size_t begin,
|
|
size_t end,
|
|
const dbc::Database *can_dbc,
|
|
bool skip_raw_can,
|
|
SeriesAccumulator *series) {
|
|
for (size_t i = begin; i < end; ++i) {
|
|
const Event &event_record = events[i];
|
|
append_event_fast(event_record.which,
|
|
event_record.eidx_segnum,
|
|
event_record.data,
|
|
can_dbc,
|
|
skip_raw_can,
|
|
0.0,
|
|
series);
|
|
}
|
|
}
|
|
|
|
void merge_route_series(RouteSeries *dst, RouteSeries *src) {
|
|
if (src->times.empty()) {
|
|
return;
|
|
}
|
|
if (dst->times.empty()) {
|
|
*dst = std::move(*src);
|
|
return;
|
|
}
|
|
|
|
dst->times.reserve(dst->times.size() + src->times.size());
|
|
dst->values.reserve(dst->values.size() + src->values.size());
|
|
dst->times.insert(dst->times.end(), src->times.begin(), src->times.end());
|
|
dst->values.insert(dst->values.end(), src->values.begin(), src->values.end());
|
|
}
|
|
|
|
void merge_can_message_data(CanMessageData *dst, CanMessageData *src) {
|
|
if (src->samples.empty()) {
|
|
return;
|
|
}
|
|
if (dst->samples.empty()) {
|
|
*dst = std::move(*src);
|
|
return;
|
|
}
|
|
dst->samples.reserve(dst->samples.size() + src->samples.size());
|
|
dst->samples.insert(dst->samples.end(),
|
|
std::make_move_iterator(src->samples.begin()),
|
|
std::make_move_iterator(src->samples.end()));
|
|
}
|
|
|
|
void merge_series_accumulator(SeriesAccumulator *dst, SeriesAccumulator *src) {
|
|
if (dst->fixed_series.size() != src->fixed_series.size()) {
|
|
throw std::runtime_error("Fixed-series slot count mismatch during merge");
|
|
}
|
|
|
|
for (size_t i = 0; i < dst->fixed_series.size(); ++i) {
|
|
merge_route_series(&dst->fixed_series[i], &src->fixed_series[i]);
|
|
}
|
|
for (auto &series : src->dynamic_series) {
|
|
if (series.path.empty()) continue;
|
|
RouteSeries &dst_series = dst->dynamic_series[ensure_dynamic_slot(series.path, dst)];
|
|
merge_route_series(&dst_series, &series);
|
|
}
|
|
for (auto &message : src->can_messages) {
|
|
CanMessageData &dst_message = *ensure_can_message(message.id.service, message.id.bus, message.id.address, dst);
|
|
merge_can_message_data(&dst_message, &message);
|
|
}
|
|
for (auto &[path, info] : src->enum_info) {
|
|
dst->enum_info.try_emplace(path, std::move(info));
|
|
}
|
|
}
|
|
|
|
size_t populated_series_count(const SeriesAccumulator &series) {
|
|
size_t count = 0;
|
|
for (const RouteSeries &fixed : series.fixed_series) {
|
|
count += !fixed.times.empty();
|
|
}
|
|
for (const RouteSeries &dynamic : series.dynamic_series) {
|
|
count += !dynamic.times.empty();
|
|
}
|
|
return count;
|
|
}
|
|
|
|
bool series_is_sorted(const RouteSeries &series) {
|
|
for (size_t i = 1; i < series.times.size(); ++i) {
|
|
if (series.times[i] < series.times[i - 1]) return false;
|
|
}
|
|
return true;
|
|
}
|
|
|
|
void sort_series_by_time(RouteSeries *series) {
|
|
if (series->times.size() <= 1 || series_is_sorted(*series)) {
|
|
return;
|
|
}
|
|
std::vector<size_t> order(series->times.size());
|
|
std::iota(order.begin(), order.end(), 0);
|
|
std::sort(order.begin(), order.end(), [&](size_t a, size_t b) {
|
|
return series->times[a] < series->times[b];
|
|
});
|
|
|
|
std::vector<double> sorted_times(series->times.size());
|
|
std::vector<double> sorted_values(series->values.size());
|
|
for (size_t i = 0; i < order.size(); ++i) {
|
|
sorted_times[i] = series->times[order[i]];
|
|
sorted_values[i] = series->values[order[i]];
|
|
}
|
|
series->times = std::move(sorted_times);
|
|
series->values = std::move(sorted_values);
|
|
}
|
|
|
|
std::vector<RouteSeries> collect_series(SeriesAccumulator &&series) {
|
|
std::vector<RouteSeries> out;
|
|
out.reserve(series.fixed_series.size() + series.dynamic_series.size());
|
|
for (auto &fixed : series.fixed_series) {
|
|
sort_series_by_time(&fixed);
|
|
if (!fixed.times.empty()) {
|
|
out.push_back(std::move(fixed));
|
|
}
|
|
}
|
|
for (auto &dynamic : series.dynamic_series) {
|
|
sort_series_by_time(&dynamic);
|
|
if (!dynamic.times.empty()) {
|
|
out.push_back(std::move(dynamic));
|
|
}
|
|
}
|
|
return out;
|
|
}
|
|
|
|
RouteData build_route_data(std::vector<RouteSeries> &&series_list,
|
|
std::vector<CanMessageData> &&can_messages,
|
|
std::vector<LogEntry> &&logs,
|
|
std::vector<TimelineEntry> &&timeline,
|
|
std::unordered_map<std::string, EnumInfo> &&enum_info,
|
|
std::string car_fingerprint,
|
|
std::string dbc_name) {
|
|
RouteData route_data;
|
|
route_data.series.reserve(series_list.size());
|
|
route_data.paths.reserve(series_list.size());
|
|
for (RouteSeries &series : series_list) {
|
|
if (series.times.empty()) continue;
|
|
route_data.has_time_range = true;
|
|
route_data.x_min = route_data.series.empty() ? series.times.front() : std::min(route_data.x_min, series.times.front());
|
|
route_data.x_max = route_data.series.empty() ? series.times.back() : std::max(route_data.x_max, series.times.back());
|
|
route_data.paths.push_back(series.path);
|
|
route_data.series.push_back(std::move(series));
|
|
}
|
|
|
|
std::sort(route_data.paths.begin(), route_data.paths.end());
|
|
std::sort(route_data.series.begin(), route_data.series.end(), [](const RouteSeries &a, const RouteSeries &b) {
|
|
return a.path < b.path;
|
|
});
|
|
std::sort(logs.begin(), logs.end(), [](const LogEntry &a, const LogEntry &b) {
|
|
return a.mono_time < b.mono_time;
|
|
});
|
|
logs.erase(std::unique(logs.begin(), logs.end(), [](const LogEntry &a, const LogEntry &b) {
|
|
return same_log_entry(a, b);
|
|
}),
|
|
logs.end());
|
|
|
|
std::vector<LogEntry> deduped_logs;
|
|
deduped_logs.reserve(logs.size());
|
|
for (LogEntry &entry : logs) {
|
|
if (!deduped_logs.empty()
|
|
&& entry.origin == LogOrigin::Alert
|
|
&& deduped_logs.back().origin == LogOrigin::Alert
|
|
&& deduped_logs.back().func == entry.func
|
|
&& deduped_logs.back().message == entry.message) {
|
|
continue;
|
|
}
|
|
deduped_logs.push_back(std::move(entry));
|
|
}
|
|
route_data.logs = std::move(deduped_logs);
|
|
|
|
if (!route_data.has_time_range && !route_data.logs.empty()) {
|
|
route_data.has_time_range = true;
|
|
route_data.x_min = route_data.logs.front().mono_time;
|
|
route_data.x_max = route_data.logs.back().mono_time;
|
|
}
|
|
if (!route_data.has_time_range) {
|
|
bool initialized = false;
|
|
for (const CanMessageData &message : can_messages) {
|
|
if (message.samples.empty()) continue;
|
|
if (!initialized) {
|
|
route_data.x_min = message.samples.front().mono_time;
|
|
route_data.x_max = message.samples.back().mono_time;
|
|
initialized = true;
|
|
} else {
|
|
route_data.x_min = std::min(route_data.x_min, message.samples.front().mono_time);
|
|
route_data.x_max = std::max(route_data.x_max, message.samples.back().mono_time);
|
|
}
|
|
}
|
|
route_data.has_time_range = initialized;
|
|
}
|
|
if (!route_data.has_time_range && !timeline.empty()) {
|
|
route_data.has_time_range = true;
|
|
route_data.x_min = timeline.front().start_time;
|
|
route_data.x_max = timeline.back().end_time;
|
|
}
|
|
|
|
if (route_data.has_time_range) {
|
|
const double time_offset = route_data.x_min;
|
|
for (RouteSeries &series : route_data.series) {
|
|
for (double &tm : series.times) {
|
|
tm -= time_offset;
|
|
}
|
|
}
|
|
for (LogEntry &entry : route_data.logs) {
|
|
entry.boot_time = entry.mono_time;
|
|
entry.mono_time -= time_offset;
|
|
}
|
|
for (CanMessageData &message : can_messages) {
|
|
for (CanFrameSample &sample : message.samples) {
|
|
sample.mono_time -= time_offset;
|
|
}
|
|
}
|
|
for (TimelineEntry &entry : timeline) {
|
|
entry.start_time -= time_offset;
|
|
entry.end_time -= time_offset;
|
|
}
|
|
route_data.x_max -= time_offset;
|
|
route_data.x_min = 0.0;
|
|
}
|
|
|
|
std::sort(timeline.begin(), timeline.end(), [](const TimelineEntry &a, const TimelineEntry &b) {
|
|
return a.start_time < b.start_time;
|
|
});
|
|
std::vector<TimelineEntry> merged_timeline;
|
|
merged_timeline.reserve(timeline.size());
|
|
for (TimelineEntry &entry : timeline) {
|
|
if (!merged_timeline.empty() && merged_timeline.back().type == entry.type) {
|
|
merged_timeline.back().end_time = std::max(merged_timeline.back().end_time, entry.end_time);
|
|
continue;
|
|
}
|
|
merged_timeline.push_back(std::move(entry));
|
|
}
|
|
route_data.timeline = std::move(merged_timeline);
|
|
std::sort(can_messages.begin(), can_messages.end(), [](const CanMessageData &a, const CanMessageData &b) {
|
|
return std::make_tuple(a.id.service, a.id.bus, a.id.address)
|
|
< std::make_tuple(b.id.service, b.id.bus, b.id.address);
|
|
});
|
|
route_data.can_messages = std::move(can_messages);
|
|
|
|
route_data.enum_info = std::move(enum_info);
|
|
route_data.car_fingerprint = std::move(car_fingerprint);
|
|
route_data.dbc_name = std::move(dbc_name);
|
|
rebuild_gps_trace(&route_data);
|
|
route_data.roots = collect_route_roots_for_paths(route_data.paths);
|
|
return route_data;
|
|
}
|
|
|
|
const RouteSeries *find_route_series(const RouteData &route_data, std::string_view path) {
|
|
auto it = std::lower_bound(route_data.series.begin(), route_data.series.end(), path,
|
|
[](const RouteSeries &series, std::string_view target) {
|
|
return series.path < target;
|
|
});
|
|
if (it == route_data.series.end() || it->path != path) return nullptr;
|
|
return &(*it);
|
|
}
|
|
|
|
std::optional<double> sample_series_at_time(const RouteSeries &series, double tm) {
|
|
if (series.times.empty() || series.times.size() != series.values.size()) {
|
|
return std::nullopt;
|
|
}
|
|
if (tm <= series.times.front()) {
|
|
return series.values.front();
|
|
}
|
|
if (tm >= series.times.back()) {
|
|
return series.values.back();
|
|
}
|
|
auto upper = std::lower_bound(series.times.begin(), series.times.end(), tm);
|
|
if (upper == series.times.begin()) {
|
|
return series.values.front();
|
|
}
|
|
if (upper == series.times.end()) {
|
|
return series.values.back();
|
|
}
|
|
const size_t upper_index = static_cast<size_t>(std::distance(series.times.begin(), upper));
|
|
const size_t lower_index = upper_index - 1;
|
|
const double t0 = series.times[lower_index];
|
|
const double t1 = series.times[upper_index];
|
|
const double v0 = series.values[lower_index];
|
|
const double v1 = series.values[upper_index];
|
|
if (t1 <= t0) {
|
|
return v0;
|
|
}
|
|
const double alpha = (tm - t0) / (t1 - t0);
|
|
return v0 + (v1 - v0) * alpha;
|
|
}
|
|
|
|
} // namespace
|
|
|
|
void rebuild_gps_trace(RouteData *route_data) {
|
|
route_data->gps_trace = {};
|
|
const RouteSeries *latitude = find_route_series(*route_data, "/gpsLocationExternal/latitude");
|
|
const RouteSeries *longitude = find_route_series(*route_data, "/gpsLocationExternal/longitude");
|
|
const RouteSeries *has_fix = find_route_series(*route_data, "/gpsLocationExternal/hasFix");
|
|
if (latitude == nullptr || longitude == nullptr || has_fix == nullptr) {
|
|
return;
|
|
}
|
|
|
|
const RouteSeries *bearing = find_route_series(*route_data, "/gpsLocationExternal/bearingDeg");
|
|
size_t count = std::min({latitude->times.size(), latitude->values.size(),
|
|
longitude->times.size(), longitude->values.size(),
|
|
has_fix->times.size(), has_fix->values.size()});
|
|
if (count == 0) {
|
|
return;
|
|
}
|
|
|
|
bool found = false;
|
|
route_data->gps_trace.points.reserve(count);
|
|
for (size_t i = 0; i < count; ++i) {
|
|
if (has_fix->values[i] < 0.5) {
|
|
continue;
|
|
}
|
|
const double lat = latitude->values[i];
|
|
const double lon = longitude->values[i];
|
|
if (!std::isfinite(lat) || !std::isfinite(lon)) {
|
|
continue;
|
|
}
|
|
const double tm = latitude->times[i];
|
|
const float bearing_value = bearing != nullptr
|
|
? static_cast<float>(sample_series_at_time(*bearing, tm).value_or(0.0))
|
|
: 0.0f;
|
|
route_data->gps_trace.points.push_back(GpsPoint{
|
|
.time = tm,
|
|
.lat = lat,
|
|
.lon = lon,
|
|
.bearing = bearing_value,
|
|
.type = timeline_type_at_time(route_data->timeline, tm),
|
|
});
|
|
if (!found) {
|
|
route_data->gps_trace.min_lat = route_data->gps_trace.max_lat = lat;
|
|
route_data->gps_trace.min_lon = route_data->gps_trace.max_lon = lon;
|
|
found = true;
|
|
} else {
|
|
route_data->gps_trace.min_lat = std::min(route_data->gps_trace.min_lat, lat);
|
|
route_data->gps_trace.max_lat = std::max(route_data->gps_trace.max_lat, lat);
|
|
route_data->gps_trace.min_lon = std::min(route_data->gps_trace.min_lon, lon);
|
|
route_data->gps_trace.max_lon = std::max(route_data->gps_trace.max_lon, lon);
|
|
}
|
|
}
|
|
if (!found) {
|
|
route_data->gps_trace = {};
|
|
}
|
|
}
|
|
|
|
namespace {
|
|
|
|
void build_camera_index(const std::map<int, SegmentLogs> &segments,
|
|
const RouteData &route_data,
|
|
std::string SegmentLogs::*file_member,
|
|
std::string_view index_name,
|
|
CameraFeedIndex *out) {
|
|
*out = {};
|
|
out->segment_files.reserve(segments.size());
|
|
|
|
std::unordered_set<int> available_segments;
|
|
available_segments.reserve(segments.size());
|
|
for (const auto &[segment_number, segment] : segments) {
|
|
const std::string &path = segment.*file_member;
|
|
if (path.empty()) continue;
|
|
out->segment_files.push_back(CameraSegmentFile{
|
|
.segment = segment_number,
|
|
.path = path,
|
|
});
|
|
available_segments.insert(segment_number);
|
|
}
|
|
if (out->segment_files.empty()) {
|
|
return;
|
|
}
|
|
|
|
const std::string prefix = "/" + std::string(index_name);
|
|
const RouteSeries *segment_numbers = find_route_series(route_data, prefix + "/segmentNum");
|
|
const RouteSeries *decode_indices = find_route_series(route_data, prefix + "/segmentId");
|
|
if (decode_indices == nullptr) {
|
|
decode_indices = find_route_series(route_data, prefix + "/segmentIdEncode");
|
|
}
|
|
const RouteSeries *frame_ids = find_route_series(route_data, prefix + "/frameId");
|
|
if (segment_numbers == nullptr || decode_indices == nullptr) {
|
|
return;
|
|
}
|
|
|
|
size_t count = std::min(segment_numbers->times.size(), segment_numbers->values.size());
|
|
count = std::min(count, decode_indices->values.size());
|
|
if (frame_ids != nullptr) {
|
|
count = std::min(count, frame_ids->values.size());
|
|
}
|
|
out->entries.reserve(count);
|
|
for (size_t i = 0; i < count; ++i) {
|
|
const int segment_number = static_cast<int>(std::llround(segment_numbers->values[i]));
|
|
if (available_segments.find(segment_number) == available_segments.end()) {
|
|
continue;
|
|
}
|
|
const int decode_index = static_cast<int>(std::llround(decode_indices->values[i]));
|
|
const uint32_t frame_id = frame_ids != nullptr
|
|
? static_cast<uint32_t>(std::llround(frame_ids->values[i]))
|
|
: static_cast<uint32_t>(std::max(0, decode_index));
|
|
out->entries.push_back(CameraFrameIndexEntry{
|
|
.timestamp = segment_numbers->times[i],
|
|
.segment = segment_number,
|
|
.decode_index = decode_index,
|
|
.frame_id = frame_id,
|
|
});
|
|
}
|
|
|
|
std::sort(out->entries.begin(), out->entries.end(),
|
|
[](const CameraFrameIndexEntry &a, const CameraFrameIndexEntry &b) {
|
|
return a.timestamp < b.timestamp;
|
|
});
|
|
}
|
|
|
|
size_t load_worker_budget() {
|
|
size_t worker_count = std::thread::hardware_concurrency();
|
|
if (worker_count == 0) {
|
|
worker_count = 1;
|
|
}
|
|
if (const char *raw = std::getenv("JOTP_LOAD_WORKERS"); raw != nullptr && std::strlen(raw) > 0) {
|
|
char *end = nullptr;
|
|
const unsigned long parsed = std::strtoul(raw, &end, 10);
|
|
if (end != nullptr && *end == '\0' && parsed > 0) {
|
|
worker_count = static_cast<size_t>(parsed);
|
|
}
|
|
}
|
|
return std::max<size_t>(1, worker_count);
|
|
}
|
|
|
|
size_t segment_worker_count(size_t segment_count, size_t worker_budget) {
|
|
return std::max<size_t>(1, std::min(worker_budget, segment_count));
|
|
}
|
|
|
|
size_t extract_chunk_count(size_t event_count, size_t worker_budget, size_t segment_workers) {
|
|
if (event_count < 4096) return 1;
|
|
const size_t per_segment_budget = std::max<size_t>(1, worker_budget / std::max<size_t>(1, segment_workers));
|
|
const size_t chunk_target = std::max<size_t>(1, (event_count + 14999) / 15000);
|
|
return std::max<size_t>(1, std::min({per_segment_budget, chunk_target, static_cast<size_t>(8)}));
|
|
}
|
|
|
|
SeriesAccumulator extract_segment_series(const std::vector<Event> &events,
|
|
const SchemaIndex &schema,
|
|
const dbc::Database *can_dbc,
|
|
bool skip_raw_can,
|
|
size_t worker_budget,
|
|
size_t segment_workers) {
|
|
const size_t chunk_count = extract_chunk_count(events.size(), worker_budget, segment_workers);
|
|
if (chunk_count <= 1 || events.empty()) {
|
|
SeriesAccumulator series = make_series_accumulator(schema);
|
|
append_events_fast_range(events, 0, events.size(), can_dbc, skip_raw_can, &series);
|
|
return series;
|
|
}
|
|
|
|
const size_t events_per_chunk = (events.size() + chunk_count - 1) / chunk_count;
|
|
std::vector<SeriesAccumulator> chunk_results;
|
|
chunk_results.reserve(chunk_count);
|
|
for (size_t i = 0; i < chunk_count; ++i) {
|
|
chunk_results.push_back(make_series_accumulator(schema));
|
|
}
|
|
|
|
std::vector<std::thread> workers;
|
|
workers.reserve(chunk_count > 0 ? chunk_count - 1 : 0);
|
|
for (size_t chunk = 1; chunk < chunk_count; ++chunk) {
|
|
workers.emplace_back([&, chunk]() {
|
|
const size_t begin = chunk * events_per_chunk;
|
|
const size_t end = std::min(events.size(), begin + events_per_chunk);
|
|
append_events_fast_range(events, begin, end, can_dbc, skip_raw_can, &chunk_results[chunk]);
|
|
});
|
|
}
|
|
append_events_fast_range(events, 0, std::min(events.size(), events_per_chunk), can_dbc, skip_raw_can, &chunk_results[0]);
|
|
for (std::thread &worker : workers) {
|
|
worker.join();
|
|
}
|
|
|
|
SeriesAccumulator merged = make_series_accumulator(schema);
|
|
for (SeriesAccumulator &chunk : chunk_results) {
|
|
merge_series_accumulator(&merged, &chunk);
|
|
}
|
|
return merged;
|
|
}
|
|
|
|
LoadedRouteArtifacts load_route_series_parallel(
|
|
const std::map<int, SegmentLogs> &segments,
|
|
const SchemaIndex &schema,
|
|
const dbc::Database *can_dbc,
|
|
LogSelector selector,
|
|
bool skip_raw_can,
|
|
LoadStats *stats) {
|
|
struct SegmentResult {
|
|
SeriesAccumulator series;
|
|
std::vector<LogEntry> logs;
|
|
std::vector<TimelineEntry> timeline;
|
|
};
|
|
|
|
const std::vector<std::pair<int, SegmentLogs>> segment_list(segments.begin(), segments.end());
|
|
std::vector<SegmentResult> results;
|
|
results.reserve(segment_list.size());
|
|
for (size_t i = 0; i < segment_list.size(); ++i) {
|
|
results.emplace_back(SegmentResult{make_series_accumulator(schema)});
|
|
}
|
|
std::atomic<size_t> next_segment{0};
|
|
std::mutex error_mutex;
|
|
std::string first_error;
|
|
const size_t worker_budget = static_cast<size_t>(stats->num_workers);
|
|
const size_t segment_workers = segment_worker_count(segment_list.size(), worker_budget);
|
|
|
|
auto worker = [&]() {
|
|
while (true) {
|
|
const size_t index = next_segment.fetch_add(1);
|
|
if (index >= segment_list.size()) {
|
|
return;
|
|
}
|
|
|
|
const auto &[segment_number, segment] = segment_list[index];
|
|
const std::string &log_path = selected_log_path(segment, selector);
|
|
LoadStats::SegmentStats &segment_stats = stats->segments[index];
|
|
segment_stats.segment_number = segment_number;
|
|
segment_stats.log_path = log_path;
|
|
if (log_path.empty()) {
|
|
segment_stats.failed = true;
|
|
std::lock_guard<std::mutex> lock(error_mutex);
|
|
if (first_error.empty()) {
|
|
first_error = "Missing log path for segment " + std::to_string(segment_number);
|
|
}
|
|
stats->publish(RouteLoadStage::DownloadingSegment, index, std::to_string(segment_number));
|
|
stats->publish(RouteLoadStage::ParsingSegment, index, std::to_string(segment_number));
|
|
continue;
|
|
}
|
|
|
|
LogReader reader;
|
|
if (!reader.load(log_path, nullptr, true)) {
|
|
segment_stats.failed = true;
|
|
std::lock_guard<std::mutex> lock(error_mutex);
|
|
if (first_error.empty()) {
|
|
first_error = "Failed to load log segment: " + log_path;
|
|
}
|
|
stats->publish(RouteLoadStage::DownloadingSegment, index, std::to_string(segment_number));
|
|
stats->publish(RouteLoadStage::ParsingSegment, index, std::to_string(segment_number));
|
|
continue;
|
|
}
|
|
|
|
segment_stats.download_seconds = reader.download_seconds();
|
|
segment_stats.decompress_seconds = reader.decompress_seconds();
|
|
segment_stats.parse_seconds = reader.parse_seconds();
|
|
segment_stats.compressed_bytes = reader.compressed_size();
|
|
segment_stats.decompressed_bytes = reader.decompressed_size();
|
|
stats->bytes_downloaded.fetch_add(reader.compressed_size());
|
|
stats->segments_downloaded.fetch_add(1);
|
|
stats->publish(RouteLoadStage::DownloadingSegment, index, std::to_string(segment_number));
|
|
|
|
const auto extract_start = LoadStats::Clock::now();
|
|
results[index].series = extract_segment_series(reader.events, schema, can_dbc, skip_raw_can, worker_budget, segment_workers);
|
|
results[index].logs = extract_segment_logs(reader.events);
|
|
results[index].timeline = extract_segment_timeline(reader.events);
|
|
segment_stats.extract_seconds = std::chrono::duration<double>(LoadStats::Clock::now() - extract_start).count();
|
|
segment_stats.event_count = reader.events.size();
|
|
segment_stats.series_count = populated_series_count(results[index].series);
|
|
stats->segments_parsed.fetch_add(1);
|
|
stats->publish(RouteLoadStage::ParsingSegment, index, std::to_string(segment_number));
|
|
}
|
|
};
|
|
|
|
std::vector<std::thread> workers;
|
|
workers.reserve(segment_workers);
|
|
for (size_t i = 0; i < segment_workers; ++i) {
|
|
workers.emplace_back(worker);
|
|
}
|
|
for (std::thread &thread : workers) {
|
|
thread.join();
|
|
}
|
|
|
|
if (!first_error.empty()) throw std::runtime_error(first_error);
|
|
|
|
stats->merge_start = LoadStats::Clock::now();
|
|
SeriesAccumulator merged = make_series_accumulator(schema);
|
|
for (size_t i = 0; i < results.size(); ++i) {
|
|
merge_series_accumulator(&merged, &results[i].series);
|
|
}
|
|
std::vector<LogEntry> logs;
|
|
std::vector<TimelineEntry> timeline;
|
|
for (SegmentResult &result : results) {
|
|
if (!result.logs.empty()) {
|
|
logs.insert(logs.end(),
|
|
std::make_move_iterator(result.logs.begin()),
|
|
std::make_move_iterator(result.logs.end()));
|
|
}
|
|
if (!result.timeline.empty()) {
|
|
timeline.insert(timeline.end(),
|
|
std::make_move_iterator(result.timeline.begin()),
|
|
std::make_move_iterator(result.timeline.end()));
|
|
}
|
|
}
|
|
LoadedRouteArtifacts artifacts;
|
|
artifacts.series = collect_series(std::move(merged));
|
|
artifacts.can_messages = std::move(merged.can_messages);
|
|
artifacts.logs = std::move(logs);
|
|
artifacts.timeline = std::move(timeline);
|
|
artifacts.enum_info = std::move(merged.enum_info);
|
|
stats->merge_end = LoadStats::Clock::now();
|
|
return artifacts;
|
|
}
|
|
|
|
std::vector<std::string> collect_layout_roots(const SketchLayout &layout) {
|
|
std::vector<std::string> roots;
|
|
for (const auto &tab : layout.tabs) {
|
|
for (const auto &pane : tab.panes) {
|
|
for (const auto &curve : pane.curves) {
|
|
std::string root = "custom";
|
|
if (is_absolute_curve(curve.name)) {
|
|
const size_t slash = curve.name.find('/', 1);
|
|
root = curve.name.substr(1, slash == std::string::npos ? std::string::npos : slash - 1);
|
|
}
|
|
if (std::find(roots.begin(), roots.end(), root) == roots.end()) {
|
|
roots.push_back(root);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if (roots.empty()) {
|
|
roots.push_back("layout");
|
|
}
|
|
return roots;
|
|
}
|
|
|
|
} // namespace
|
|
|
|
std::vector<std::string> collect_route_roots_for_paths(const std::vector<std::string> &paths) {
|
|
std::vector<std::string> roots;
|
|
for (const std::string &path : paths) {
|
|
if (!is_absolute_curve(path)) continue;
|
|
const size_t slash = path.find('/', 1);
|
|
const std::string root = path.substr(1, slash == std::string::npos ? std::string::npos : slash - 1);
|
|
if (!root.empty() && std::find(roots.begin(), roots.end(), root) == roots.end()) {
|
|
roots.push_back(root);
|
|
}
|
|
}
|
|
std::sort(roots.begin(), roots.end());
|
|
return roots;
|
|
}
|
|
|
|
struct StreamAccumulator::Impl {
|
|
const SchemaIndex &schema = SchemaIndex::instance();
|
|
SeriesAccumulator series = make_series_accumulator(schema);
|
|
std::vector<LogEntry> logs;
|
|
std::vector<TimelineEntry> timeline;
|
|
std::string last_alert_key;
|
|
std::string manual_dbc_name;
|
|
std::string detected_dbc_name;
|
|
std::string car_fingerprint;
|
|
std::optional<dbc::Database> can_dbc;
|
|
std::optional<double> time_offset;
|
|
|
|
void refresh_dbc() {
|
|
const std::string next_dbc = !manual_dbc_name.empty() ? manual_dbc_name : detect_dbc_for_fingerprint(car_fingerprint);
|
|
if (next_dbc == detected_dbc_name) {
|
|
return;
|
|
}
|
|
detected_dbc_name = next_dbc;
|
|
can_dbc = load_dbc_by_name(detected_dbc_name);
|
|
}
|
|
};
|
|
|
|
StreamAccumulator::StreamAccumulator(const std::string &dbc_name, std::optional<double> time_offset)
|
|
: impl_(std::make_unique<Impl>()) {
|
|
impl_->manual_dbc_name = dbc_name;
|
|
impl_->time_offset = time_offset;
|
|
impl_->refresh_dbc();
|
|
}
|
|
|
|
StreamAccumulator::~StreamAccumulator() = default;
|
|
|
|
void StreamAccumulator::setDbcName(const std::string &dbc_name) {
|
|
impl_->manual_dbc_name = dbc_name;
|
|
impl_->refresh_dbc();
|
|
}
|
|
|
|
void StreamAccumulator::appendEvent(kj::ArrayPtr<const capnp::word> data) {
|
|
with_parseable_event(data, [&](const cereal::Event::Reader &event) {
|
|
const cereal::Event::Which which = event.which();
|
|
const double boot_time = static_cast<double>(event.getLogMonoTime()) / 1.0e9;
|
|
if (!impl_->time_offset.has_value()) {
|
|
impl_->time_offset = boot_time;
|
|
}
|
|
if (which == cereal::Event::Which::CAR_PARAMS) {
|
|
const std::string fingerprint = event.getCarParams().getCarFingerprint().cStr();
|
|
if (!fingerprint.empty() && fingerprint != impl_->car_fingerprint) {
|
|
impl_->car_fingerprint = fingerprint;
|
|
impl_->refresh_dbc();
|
|
}
|
|
}
|
|
|
|
append_event_static_reader(which,
|
|
event,
|
|
impl_->can_dbc ? &*impl_->can_dbc : nullptr,
|
|
impl_->can_dbc.has_value(),
|
|
*impl_->time_offset,
|
|
&impl_->series);
|
|
append_log_event(which, event, *impl_->time_offset, &impl_->logs, &impl_->last_alert_key);
|
|
if (which == cereal::Event::Which::SELFDRIVE_STATE) {
|
|
const auto sd = event.getSelfdriveState();
|
|
append_timeline_entry(&impl_->timeline, boot_time - *impl_->time_offset,
|
|
alert_status_to_timeline_type(sd.getAlertStatus(), sd.getEnabled()));
|
|
}
|
|
});
|
|
}
|
|
|
|
void StreamAccumulator::appendCanFrames(CanServiceKind service, const std::vector<LiveCanFrame> &frames) {
|
|
if (frames.empty()) {
|
|
return;
|
|
}
|
|
if (!impl_->time_offset.has_value()) {
|
|
impl_->time_offset = frames.front().mono_time;
|
|
}
|
|
for (const LiveCanFrame &frame : frames) {
|
|
append_live_can_frame(service,
|
|
frame,
|
|
*impl_->time_offset,
|
|
impl_->can_dbc ? &*impl_->can_dbc : nullptr,
|
|
&impl_->series);
|
|
}
|
|
}
|
|
|
|
StreamExtractBatch StreamAccumulator::takeBatch() {
|
|
StreamExtractBatch batch;
|
|
batch.car_fingerprint = impl_->car_fingerprint;
|
|
batch.dbc_name = impl_->detected_dbc_name;
|
|
if (impl_->time_offset.has_value()) {
|
|
batch.has_time_offset = true;
|
|
batch.time_offset = *impl_->time_offset;
|
|
}
|
|
if (impl_->logs.empty() && impl_->timeline.empty()
|
|
&& populated_series_count(impl_->series) == 0
|
|
&& impl_->series.enum_info.empty()
|
|
&& impl_->series.can_messages.empty()) {
|
|
return batch;
|
|
}
|
|
|
|
SeriesAccumulator emitted = std::move(impl_->series);
|
|
batch.can_messages = std::move(emitted.can_messages);
|
|
batch.enum_info = std::move(emitted.enum_info);
|
|
batch.series = collect_series(std::move(emitted));
|
|
batch.logs = std::move(impl_->logs);
|
|
batch.timeline = std::move(impl_->timeline);
|
|
impl_->series = make_series_accumulator(impl_->schema);
|
|
impl_->logs.clear();
|
|
impl_->timeline.clear();
|
|
return batch;
|
|
}
|
|
|
|
const std::string &StreamAccumulator::carFingerprint() const {
|
|
return impl_->car_fingerprint;
|
|
}
|
|
|
|
const std::string &StreamAccumulator::dbc_name() const {
|
|
return impl_->detected_dbc_name;
|
|
}
|
|
|
|
std::optional<double> StreamAccumulator::timeOffset() const {
|
|
return impl_->time_offset;
|
|
}
|
|
|
|
SketchLayout load_sketch_layout(const fs::path &layout_path) {
|
|
SketchLayout layout = parse_layout(layout_path);
|
|
layout.roots = collect_layout_roots(layout);
|
|
return layout;
|
|
}
|
|
|
|
RouteData load_route_data(const std::string &route_name,
|
|
const std::string &data_dir,
|
|
const std::string &dbc_name,
|
|
const RouteLoadProgressCallback &progress) {
|
|
if (route_name.empty()) return RouteData{};
|
|
|
|
const RouteSelection route = parse_route_selection(route_name);
|
|
if (route.canonical_name.empty() || (data_dir.empty() && route.dongle_id.empty())) {
|
|
throw std::runtime_error("Invalid route format: " + route_name);
|
|
}
|
|
LoadStats stats(progress);
|
|
stats.load_start = LoadStats::Clock::now();
|
|
std::map<int, SegmentLogs> segments = data_dir.empty()
|
|
? load_segments_from_server(route)
|
|
: load_segments_from_local(route, data_dir);
|
|
segments = trim_segments(std::move(segments), route);
|
|
if (segments.empty()) throw std::runtime_error("No log segments found for " + route_name);
|
|
stats.resolve_end = LoadStats::Clock::now();
|
|
stats.segment_count = segments.size();
|
|
stats.total_segments.store(segments.size());
|
|
stats.num_workers = static_cast<int>(load_worker_budget());
|
|
stats.segments.resize(segments.size());
|
|
stats.publish(RouteLoadStage::Resolving, 0, {});
|
|
|
|
const RouteMetadata metadata = detect_route_metadata(segments, route.selector);
|
|
const std::string resolved_dbc = !dbc_name.empty() ? dbc_name : detect_dbc_for_fingerprint(metadata.car_fingerprint);
|
|
const std::optional<dbc::Database> can_dbc = load_dbc_by_name(resolved_dbc);
|
|
|
|
const SchemaIndex &schema = SchemaIndex::instance();
|
|
LoadedRouteArtifacts artifacts = load_route_series_parallel(segments, schema, can_dbc ? &*can_dbc : nullptr,
|
|
route.selector, can_dbc.has_value(), &stats);
|
|
RouteData route_data = build_route_data(std::move(artifacts.series),
|
|
std::move(artifacts.can_messages),
|
|
std::move(artifacts.logs),
|
|
std::move(artifacts.timeline),
|
|
std::move(artifacts.enum_info),
|
|
metadata.car_fingerprint,
|
|
resolved_dbc);
|
|
route_data.route_id = make_route_identifier(route, segments);
|
|
build_camera_index(segments, route_data, &SegmentLogs::fcamera, "roadEncodeIdx", &route_data.road_camera);
|
|
build_camera_index(segments, route_data, &SegmentLogs::dcamera, "driverEncodeIdx", &route_data.driver_camera);
|
|
build_camera_index(segments, route_data, &SegmentLogs::ecamera, "wideRoadEncodeIdx", &route_data.wide_road_camera);
|
|
build_camera_index(segments, route_data, &SegmentLogs::qcamera, "qRoadEncodeIdx", &route_data.qroad_camera);
|
|
stats.load_end = LoadStats::Clock::now();
|
|
stats.publish(RouteLoadStage::Finished, segments.size(), {});
|
|
stats.print_summary(route_data.series.size());
|
|
return route_data;
|
|
}
|
|
|
|
RouteIdentifier parse_route_identifier(std::string_view route_name) {
|
|
return make_route_identifier(parse_route_selection(std::string(route_name)), {});
|
|
}
|
|
|
|
std::vector<std::string> available_dbc_names() {
|
|
return available_dbc_names_impl();
|
|
}
|
|
|
|
std::optional<dbc::Database> load_dbc_by_name(const std::string &dbc_name) {
|
|
if (dbc_name.empty()) {
|
|
return std::nullopt;
|
|
}
|
|
try {
|
|
return std::optional<dbc::Database>(std::in_place, resolve_dbc_path(dbc_name));
|
|
} catch (...) {
|
|
return std::nullopt;
|
|
}
|
|
}
|
|
|
|
std::vector<RouteSeries> decode_can_messages(const std::vector<CanMessageData> &can_messages,
|
|
const std::string &dbc_name,
|
|
std::unordered_map<std::string, EnumInfo> *enum_info) {
|
|
if (enum_info != nullptr) {
|
|
enum_info->clear();
|
|
}
|
|
const std::optional<dbc::Database> can_dbc = load_dbc_by_name(dbc_name);
|
|
if (!can_dbc.has_value()) {
|
|
return {};
|
|
}
|
|
|
|
SeriesAccumulator series;
|
|
for (const CanMessageData &message : can_messages) {
|
|
const char *service_name = message.id.service == CanServiceKind::Can ? "can" : "sendcan";
|
|
for (const CanFrameSample &sample : message.samples) {
|
|
decode_can_frame(&*can_dbc,
|
|
service_name,
|
|
message.id.bus,
|
|
message.id.address,
|
|
reinterpret_cast<const uint8_t *>(sample.data.data()),
|
|
sample.data.size(),
|
|
sample.mono_time,
|
|
&series);
|
|
}
|
|
}
|
|
if (enum_info != nullptr) {
|
|
*enum_info = std::move(series.enum_info);
|
|
}
|
|
return collect_series(std::move(series));
|
|
}
|