diff --git a/tools/jotpluggler/.gitignore b/tools/jotpluggler/.gitignore index 7cb98300f..5f5f9dbde 100644 --- a/tools/jotpluggler/.gitignore +++ b/tools/jotpluggler/.gitignore @@ -3,6 +3,7 @@ jot_*.o *.o jotpluggler car_fingerprint_to_dbc.h +generated_event_extractors.h generated_dbcs/.stamp generated_dbcs/*.dbc layouts/.jotpluggler_autosave/ diff --git a/tools/jotpluggler/SConscript b/tools/jotpluggler/SConscript index 122d50234..082d05244 100644 --- a/tools/jotpluggler/SConscript +++ b/tools/jotpluggler/SConscript @@ -1,4 +1,5 @@ import os +import subprocess import imgui import libusb from opendbc import get_generated_dbcs @@ -77,8 +78,24 @@ def write_car_fingerprint_to_dbc_header(target, source, env): return None +def generate_event_extractors(target, source, env): + subprocess.check_call([ + "python3", + "tools/jotpluggler/generate_event_extractors.py", + os.path.realpath(BASEDIR), + str(target[0]), + ]) + return None + generated_dbc_stamp = jot_env.Command(f"generated_dbcs/.stamp", [], materialize_generated_dbcs) car_fingerprint_to_dbc = jot_env.Command("car_fingerprint_to_dbc.h", [], write_car_fingerprint_to_dbc_header) +event_extractors = jot_env.Command("generated_event_extractors.h", [ + "generate_event_extractors.py", + jot_env.Glob("#cereal/*.capnp"), + jot_env.Glob("#cereal/include/*.capnp"), + ], + generate_event_extractors, +) libs = [replay_lib, common, messaging, visionipc, cereal, File(f"{imgui.LIB_DIR}/libimgui.a"), File(f"{imgui.LIB_DIR}/libglfw3.a"), "avformat", "avcodec", "avutil", "x264", "yuv", "z", "bz2", "zstd", "m", "pthread", "usb-1.0"] @@ -90,3 +107,4 @@ else: program = jot_env.Program("jotpluggler", jot_env.Glob("*.cc"), LIBS=libs) jot_env.Depends(program, generated_dbc_stamp) jot_env.Depends(program, car_fingerprint_to_dbc) +jot_env.Depends(program, event_extractors) diff --git a/tools/jotpluggler/generate_event_extractors.py b/tools/jotpluggler/generate_event_extractors.py new file mode 100644 index 000000000..1aba9578c --- /dev/null +++ b/tools/jotpluggler/generate_event_extractors.py @@ -0,0 +1,327 @@ +import sys +import capnp +from pathlib import Path + +NO_DISCRIMINANT = 65535 +SCALAR_KINDS = { + "bool": "Bool", + "int8": "Int", + "int16": "Int", + "int32": "Int", + "int64": "Int", + "uint8": "UInt", + "uint16": "UInt", + "uint32": "UInt", + "uint64": "UInt", + "float32": "Float", + "float64": "Float", + "enum": "Enum", +} +NESTED_TYPE_KINDS = {"struct", "list"} +IGNORED_TYPE_KINDS = {"void", "text", "data", "interface", "anyPointer"} + + +def cxx_string(value): + return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"' + + +def accessor(prefix, name): + return prefix + name[:1].upper() + name[1:] + + +def field_type(field): + if field.proto.which() == "group": + return "struct" + return field.proto.slot.type.which() + + +def field_type_proto(field): + return field.proto.slot.type if field.proto.which() == "slot" else None + + +def scalar_kind(type_proto): + if type_proto is None: + return None + return SCALAR_KINDS.get(type_proto.which()) + + +def enum_names(schema): + if schema is None: + return [] + names_by_ordinal = schema.enumerants + if not names_by_ordinal: + return [] + max_ordinal = max(names_by_ordinal.values()) + out = [""] * (max_ordinal + 1) + for name, ordinal in names_by_ordinal.items(): + out[ordinal] = name + return out + + +class Generator: + def __init__(self, event_schema): + self.event_schema = event_schema + self.fixed_paths = [] + self.tmp_index = 0 + self.lines = [] + self.emits_memo = {} + + def tmp(self, prefix): + self.tmp_index += 1 + return f"{prefix}_{self.tmp_index}" + + def add_fixed_path(self, path): + slot = len(self.fixed_paths) + self.fixed_paths.append(path) + return slot + + def emit(self, indent, text=""): + self.lines.append(" " * indent + text) + + def scalar_double_expr(self, value_expr, kind): + if kind == "Bool": + return f"({value_expr} ? 1.0 : 0.0)" + if kind == "Enum": + return f"static_cast(static_cast({value_expr}))" + return f"static_cast({value_expr})" + + def emit_enum_capture(self, indent, path_expr, names): + if not names: + return + names_expr = "{" + ", ".join(cxx_string(name) for name in names) + "}" + self.emit(indent, f"capture_static_enum_info({path_expr}, {names_expr}, series);") + + def emit_node(self, indent, type_kind, type_proto, schema, expr, path, path_expr, dynamic_path): + if not self.node_emits(type_kind, type_proto, schema): + return + kind = scalar_kind(type_proto) + if kind is not None: + double_expr = self.scalar_double_expr(expr, kind) + if dynamic_path: + if kind == "Enum": + self.emit_enum_capture(indent, path_expr, enum_names(schema)) + self.emit(indent, f"append_dynamic_scalar_point({path_expr}, tm, {double_expr}, series);") + else: + slot = self.add_fixed_path(path) + if kind == "Enum": + self.emit_enum_capture(indent, cxx_string(path), enum_names(schema)) + self.emit(indent, f"append_fixed_scalar_point(&series->fixed_series[{slot}], tm, {double_expr});") + return + + if type_kind == "struct": + self.emit_struct(indent, schema, expr, path, path_expr, dynamic_path) + return + + if type_kind == "list": + self.emit_list(indent, type_proto, schema, expr, path, path_expr, dynamic_path) + + def emit_field(self, indent, struct_schema, reader_expr, field_name, base_path, base_path_expr, dynamic_path): + field = struct_schema.fields[field_name] + proto = field.proto + type_kind = field_type(field) + type_proto = field_type_proto(field) + kind = scalar_kind(type_proto) + value_schema = field.schema if kind == "Enum" or type_kind in NESTED_TYPE_KINDS else None + if not self.node_emits(type_kind, type_proto, value_schema): + return + + field_path = f"{base_path}/{field_name}" + field_path_expr = None + if dynamic_path: + field_path_var = self.tmp("path") + self.emit(indent, f"const std::string {field_path_var} = {base_path_expr} + {cxx_string('/' + field_name)};") + field_path_expr = field_path_var + + get_call = f"{reader_expr}.{accessor('get', field_name)}()" + has_call = f"{reader_expr}.{accessor('has', field_name)}()" + conditions = [] + if proto.discriminantValue != NO_DISCRIMINANT: + conditions.append(f"{reader_expr}.which() == static_cast({proto.discriminantValue})") + if proto.which() == "slot" and type_kind in NESTED_TYPE_KINDS: + conditions.append(has_call) + + if conditions: + self.emit(indent, f"if ({' && '.join(conditions)}) {{") + indent += 2 + + value_var = self.tmp("value") + self.emit(indent, f"const auto {value_var} = {get_call};") + self.emit_node(indent, type_kind, type_proto, value_schema, value_var, field_path, field_path_expr, dynamic_path) + + if conditions: + indent -= 2 + self.emit(indent, "}") + + def emit_struct(self, indent, schema, reader_expr, path, path_expr, dynamic_path): + if schema is None: + return + for field_name in schema.fieldnames: + self.emit_field(indent, schema, reader_expr, field_name, path, path_expr, dynamic_path) + + def emit_list(self, indent, type_proto, schema, list_expr, path, path_expr, dynamic_path): + elem_type = type_proto.list.elementType + elem_kind = elem_type.which() + if elem_kind in IGNORED_TYPE_KINDS: + return + + base_path_var = path_expr + if base_path_var is None: + base_path_var = self.tmp("base_path") + self.emit(indent, f"const std::string {base_path_var} = {cxx_string(path)};") + + elem_scalar = scalar_kind(elem_type) + if elem_scalar is not None: + self.emit(indent, f"if ({list_expr}.size() <= 16) {{") + index_var = self.tmp("i") + self.emit(indent + 2, f"for (uint {index_var} = 0; {index_var} < {list_expr}.size(); ++{index_var}) {{") + item_series = self.tmp("item_series") + self.emit(indent + 4, f"RouteSeries *{item_series} = ensure_list_scalar_series({base_path_var}, {index_var}, series);") + if elem_scalar == "Enum": + self.emit_enum_capture(indent + 4, f"{item_series}->path", enum_names(schema.elementType)) + self.emit(indent + 4, f"append_fixed_scalar_point({item_series}, tm, {self.scalar_double_expr(f'{list_expr}[{index_var}]', elem_scalar)});") + self.emit(indent + 2, "}") + self.emit(indent, "}") + return + + if elem_kind in {"struct", "list"}: + index_var = self.tmp("i") + self.emit(indent, f"for (uint {index_var} = 0; {index_var} < {list_expr}.size(); ++{index_var}) {{") + item_path = self.tmp("item_path") + self.emit(indent + 2, f"const std::string {item_path} = {base_path_var} + \"/\" + std::to_string({index_var});") + item = self.tmp("item") + self.emit(indent + 2, f"const auto {item} = {list_expr}[{index_var}];") + if elem_kind == "struct": + self.emit_struct(indent + 2, schema.elementType, item, path, item_path, True) + else: + self.emit_list(indent + 2, elem_type, schema.elementType, item, path, item_path, True) + self.emit(indent, "}") + + def node_emits(self, type_kind, type_proto, schema, seen=frozenset()): + if scalar_kind(type_proto) is not None: + return True + if type_kind == "struct": + if schema is None: + return False + schema_id = int(schema.node.id) + if schema_id in seen: + return False + if schema_id in self.emits_memo: + return self.emits_memo[schema_id] + next_seen = seen | {schema_id} + for field_name in schema.fieldnames: + field = schema.fields[field_name] + ft = field_type(field) + ftp = field_type_proto(field) + fkind = scalar_kind(ftp) + if ft in IGNORED_TYPE_KINDS: + continue + fschema = field.schema if fkind == "Enum" or ft in NESTED_TYPE_KINDS else None + if self.node_emits(ft, ftp, fschema, next_seen): + self.emits_memo[schema_id] = True + return True + self.emits_memo[schema_id] = False + return False + if type_kind == "list": + if type_proto is None or schema is None: + return False + elem_type = type_proto.list.elementType + elem_kind = elem_type.which() + if elem_kind in IGNORED_TYPE_KINDS: + return False + if scalar_kind(elem_type) is not None: + return True + if elem_kind == "struct": + return self.node_emits("struct", None, schema.elementType, seen) + if elem_kind == "list": + return self.node_emits("list", elem_type, schema.elementType, seen) + return False + + def emit_can_special(self, indent, service_name): + service_kind = "CanServiceKind::Can" if service_name == "can" else "CanServiceKind::Sendcan" + self.emit(indent, f"const CanServiceKind can_service = {service_kind};") + self.emit(indent, f"for (const auto &msg : event.{accessor('get', service_name)}()) {{") + self.emit(indent + 2, "append_can_frame(can_service, static_cast(msg.getSrc()), msg.getAddress(), msg.getDeprecated().getBusTime(), msg.getDat(), tm, series);") # noqa: E501 + self.emit(indent + 2, "if (skip_raw_can) {") + self.emit(indent + 4, "const auto dat = msg.getDat();") + self.emit(indent + 4, f"decode_can_frame(can_dbc, {cxx_string(service_name)}, static_cast(msg.getSrc()), msg.getAddress(), dat.begin(), dat.size(), tm, series);") # noqa: E501 + self.emit(indent + 2, "}") + self.emit(indent, "}") + self.emit(indent, "if (skip_raw_can) {") + self.emit(indent + 2, "return true;") + self.emit(indent, "}") + + def emit_event_case(self, field_name): + field = self.event_schema.fields[field_name] + proto = field.proto + type_kind = field_type(field) + type_proto = field_type_proto(field) + kind = scalar_kind(type_proto) + schema = field.schema if kind == "Enum" or type_kind in NESTED_TYPE_KINDS else None + self.emit(4, f"case static_cast({proto.discriminantValue}): {{") + valid_slot = self.add_fixed_path(f"/{field_name}/valid") + mono_slot = self.add_fixed_path(f"/{field_name}/logMonoTime") + seconds_slot = self.add_fixed_path(f"/{field_name}/t") + self.emit(6, f"append_fixed_scalar_point(&series->fixed_series[{valid_slot}], tm, event.getValid() ? 1.0 : 0.0);") + self.emit(6, f"append_fixed_scalar_point(&series->fixed_series[{mono_slot}], tm, static_cast(event.getLogMonoTime()));") + self.emit(6, f"append_fixed_scalar_point(&series->fixed_series[{seconds_slot}], tm, tm);") + if field_name in {"can", "sendcan"}: + self.emit_can_special(6, field_name) + if self.node_emits(type_kind, type_proto, schema): + payload = self.tmp("payload") + self.emit(6, f"const auto {payload} = event.{accessor('get', field_name)}();") + self.emit_node(6, type_kind, type_proto, schema, payload, f"/{field_name}", None, False) + self.emit(6, "return true;") + self.emit(4, "}") + + def generate(self): + self.lines = [] + self.emit(0, "// Generated by tools/jotpluggler/generate_event_extractors.py; do not edit.") + self.emit(0, "") + self.emit(0, "const std::vector &static_event_fixed_paths() {") + self.emit(2, "static const std::vector paths = {") + path_insert_at = len(self.lines) + self.emit(2, "};") + self.emit(2, "return paths;") + self.emit(0, "}") + self.emit(0, "") + self.emit(0, "void capture_static_enum_info(const std::string &path, std::initializer_list names, SeriesAccumulator *series) {") + self.emit(2, "if (series->enum_info.find(path) != series->enum_info.end()) {") + self.emit(4, "return;") + self.emit(2, "}") + self.emit(2, "EnumInfo info;") + self.emit(2, "info.names.reserve(names.size());") + self.emit(2, "for (std::string_view name : names) {") + self.emit(4, "info.names.emplace_back(name);") + self.emit(2, "}") + self.emit(2, "if (!info.names.empty()) {") + self.emit(4, "series->enum_info.emplace(path, std::move(info));") + self.emit(2, "}") + self.emit(0, "}") + self.emit(0, "") + self.emit(0, "bool append_event_static_reader(cereal::Event::Which which, const cereal::Event::Reader &event, const dbc::Database *can_dbc, bool skip_raw_can, double time_offset, SeriesAccumulator *series) {") # noqa: E501 + self.emit(2, "const double tm = static_cast(event.getLogMonoTime()) / 1.0e9 - time_offset;") + self.emit(2, "switch (which) {") + for field_name in self.event_schema.union_fields: + self.emit_event_case(field_name) + self.emit(4, "default:") + self.emit(6, "return false;") + self.emit(2, "}") + self.emit(0, "}") + + path_lines = [" " + cxx_string(path) + "," for path in self.fixed_paths] + self.lines[path_insert_at:path_insert_at] = path_lines + return "\n".join(self.lines) + "\n" + + +if __name__ == "__main__": + if len(sys.argv) != 3: + print(f"usage: {sys.argv[0]} ", file=sys.stderr) + sys.exit(2) + + repo_root = Path(sys.argv[1]).resolve() + output = Path(sys.argv[2]) + capnp.remove_import_hook() + log = capnp.load(str(repo_root / "cereal" / "log.capnp")) + generated = Generator(log.Event.schema).generate() + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(generated) diff --git a/tools/jotpluggler/sketch_layout.cc b/tools/jotpluggler/sketch_layout.cc index d9622dde6..c44a4e93c 100644 --- a/tools/jotpluggler/sketch_layout.cc +++ b/tools/jotpluggler/sketch_layout.cc @@ -2,7 +2,6 @@ #include "tools/jotpluggler/car_fingerprint_to_dbc.h" #include "tools/jotpluggler/common.h" -#include #include #include @@ -10,6 +9,7 @@ #include #include #include +#include #include #include #include @@ -51,47 +51,7 @@ struct SegmentLogs { std::string qcamera; }; -enum class ScalarKind { - None, - Bool, - Int, - UInt, - Float, - Enum, -}; - -enum class ResolvedNodeKind { - Ignore, - Scalar, - Struct, - List, -}; - -struct ResolvedNode { - ResolvedNodeKind kind = ResolvedNodeKind::Ignore; - ScalarKind scalar_kind = ScalarKind::None; - int fixed_slot = -1; - bool has_field = false; - capnp::StructSchema::Field field; - std::string segment; - std::string path; - bool skip_large_scalar_list = false; - std::vector children; - std::unique_ptr element; -}; - -struct ResolvedService { - uint16_t event_which = 0; - capnp::StructSchema::Field union_field; - std::string service_name; - int valid_slot = -1; - int log_mono_time_slot = -1; - int seconds_slot = -1; - ResolvedNode payload; -}; - struct SchemaIndex { - std::vector> by_which; size_t fixed_series_count = 0; std::vector fixed_paths; @@ -112,6 +72,27 @@ struct SeriesAccumulator { std::unordered_map 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 series; std::vector can_messages; @@ -898,125 +879,11 @@ SketchLayout parse_layout(const fs::path &layout_path) { return layout; } -ScalarKind scalar_kind_for_type(const capnp::Type &type) { - if (type.isBool()) return ScalarKind::Bool; - if (type.isInt8() || type.isInt16() || type.isInt32() || type.isInt64()) { - return ScalarKind::Int; - } - if (type.isUInt8() || type.isUInt16() || type.isUInt32() || type.isUInt64()) { - return ScalarKind::UInt; - } - if (type.isFloat32() || type.isFloat64()) { - return ScalarKind::Float; - } - if (type.isEnum()) return ScalarKind::Enum; - return ScalarKind::None; -} - -ResolvedNode build_resolved_type(const capnp::Type &type, - bool has_field, - capnp::StructSchema::Field field, - std::string segment, - std::string path, - size_t *next_fixed_slot, - std::vector *fixed_paths, - bool dynamic_path = false) { - ResolvedNode node; - node.has_field = has_field; - node.field = field; - node.segment = std::move(segment); - node.path = std::move(path); - node.scalar_kind = scalar_kind_for_type(type); - if (node.scalar_kind != ScalarKind::None) { - node.kind = ResolvedNodeKind::Scalar; - if (!dynamic_path) { - node.fixed_slot = static_cast((*next_fixed_slot)++); - fixed_paths->push_back(node.path); - } - return node; - } - - if (type.isStruct()) { - node.kind = ResolvedNodeKind::Struct; - for (auto child : type.asStruct().getFields()) { - const std::string child_segment = child.getProto().getName().cStr(); - node.children.push_back(build_resolved_type( - child.getType(), - true, - child, - child_segment, - node.path + "/" + child_segment, - next_fixed_slot, - fixed_paths, - dynamic_path)); - } - return node; - } - - if (type.isList()) { - const capnp::Type element_type = type.asList().getElementType(); - if (element_type.isText() || element_type.isData() || element_type.isInterface() || element_type.isAnyPointer()) { - node.kind = ResolvedNodeKind::Ignore; - return node; - } - node.kind = ResolvedNodeKind::List; - node.skip_large_scalar_list = scalar_kind_for_type(element_type) != ScalarKind::None; - node.element = std::make_unique( - build_resolved_type(element_type, - false, - capnp::StructSchema::Field(), - "", - node.path, - next_fixed_slot, - fixed_paths, - true)); - return node; - } - - node.kind = ResolvedNodeKind::Ignore; - return node; -} - -int register_fixed_series_path(const std::string &path, - size_t *next_fixed_slot, - std::vector *fixed_paths) { - const int slot = static_cast((*next_fixed_slot)++); - fixed_paths->push_back(path); - return slot; -} - const SchemaIndex &SchemaIndex::instance() { static const SchemaIndex index = [] { SchemaIndex out; - const auto event_schema = capnp::Schema::from().asStruct(); - uint16_t max_discriminant = 0; - for (auto union_field : event_schema.getUnionFields()) { - max_discriminant = std::max(max_discriminant, union_field.getProto().getDiscriminantValue()); - } - out.by_which.resize(static_cast(max_discriminant) + 1); - size_t next_fixed_slot = 0; - for (auto union_field : event_schema.getUnionFields()) { - ResolvedService service; - service.event_which = union_field.getProto().getDiscriminantValue(); - service.union_field = union_field; - service.service_name = union_field.getProto().getName().cStr(); - service.valid_slot = register_fixed_series_path( - "/" + service.service_name + "/valid", &next_fixed_slot, &out.fixed_paths); - service.log_mono_time_slot = register_fixed_series_path( - "/" + service.service_name + "/logMonoTime", &next_fixed_slot, &out.fixed_paths); - service.seconds_slot = register_fixed_series_path( - "/" + service.service_name + "/t", &next_fixed_slot, &out.fixed_paths); - service.payload = build_resolved_type( - union_field.getType(), - false, - capnp::StructSchema::Field(), - service.service_name, - "/" + service.service_name, - &next_fixed_slot, - &out.fixed_paths); - out.by_which[service.event_which] = std::move(service); - } - out.fixed_series_count = next_fixed_slot; + out.fixed_paths = static_event_fixed_paths(); + out.fixed_series_count = out.fixed_paths.size(); return out; }(); return index; @@ -1026,45 +893,6 @@ bool is_absolute_curve(const std::string &name) { return !name.empty() && name.front() == '/'; } -std::optional scalar_value_to_double(const capnp::DynamicValue::Reader &value, ScalarKind kind) { - switch (kind) { - case ScalarKind::Bool: - return value.as() ? 1.0 : 0.0; - case ScalarKind::Int: - return static_cast(value.as()); - case ScalarKind::UInt: - return static_cast(value.as()); - case ScalarKind::Float: - return value.as(); - case ScalarKind::Enum: - return static_cast(value.as().getRaw()); - case ScalarKind::None: - return std::nullopt; - } - return std::nullopt; -} - -void capture_enum_info(const std::string &path, - const capnp::DynamicValue::Reader &value, - SeriesAccumulator *series) { - if (series->enum_info.find(path) != series->enum_info.end()) { - return; - } - - const auto dynamic_enum = value.as(); - EnumInfo info; - for (auto enumerant : dynamic_enum.getSchema().getEnumerants()) { - const uint16_t ordinal = enumerant.getOrdinal(); - if (ordinal >= info.names.size()) { - info.names.resize(static_cast(ordinal) + 1); - } - info.names[ordinal] = enumerant.getProto().getName().cStr(); - } - if (!info.names.empty()) { - series->enum_info.emplace(path, std::move(info)); - } -} - void append_scalar_point(RouteSeries *series, const std::string &path, double tm, @@ -1195,156 +1023,9 @@ void append_dynamic_scalar_point(const std::string &path, double tm, double valu append_scalar_point(ensure_dynamic_series(path, series), path, tm, value); } -void append_scalar_value(const ResolvedNode &node, - const std::string *path_override, - const capnp::DynamicValue::Reader &raw_value, - double tm, - double value, - SeriesAccumulator *series) { - if (path_override == nullptr && node.fixed_slot >= 0) { - if (node.scalar_kind == ScalarKind::Enum) { - capture_enum_info(node.path, raw_value, series); - } - append_fixed_scalar_point(&series->fixed_series[static_cast(node.fixed_slot)], tm, value); - return; - } - - const std::string &path = path_override != nullptr ? *path_override : node.path; - if (node.scalar_kind == ScalarKind::Enum) { - capture_enum_info(path, raw_value, series); - } - append_dynamic_scalar_point(path, tm, value, series); -} - -void append_fast_node(const ResolvedNode &node, - const capnp::DynamicValue::Reader &value, - double tm, - SeriesAccumulator *series, - const std::string *path_override = nullptr) { - switch (node.kind) { - case ResolvedNodeKind::Scalar: { - if (std::optional scalar = scalar_value_to_double(value, node.scalar_kind); scalar.has_value()) { - append_scalar_value(node, path_override, value, tm, *scalar, series); - } - return; - } - case ResolvedNodeKind::Struct: { - const capnp::DynamicStruct::Reader reader = value.as(); - for (const ResolvedNode &child : node.children) { - if (!child.has_field || !reader.has(child.field)) continue; - if (path_override == nullptr) { - append_fast_node(child, reader.get(child.field), tm, series, nullptr); - } else { - const std::string child_path = child.segment.empty() ? *path_override : (*path_override + "/" + child.segment); - append_fast_node(child, reader.get(child.field), tm, series, &child_path); - } - } - return; - } - case ResolvedNodeKind::List: { - if (!node.element) { - return; - } - const capnp::DynamicList::Reader list = value.as(); - if (list.size() == 0) { - return; - } - if (node.skip_large_scalar_list && list.size() > 16) { - return; - } - const std::string &base_path = path_override != nullptr ? *path_override : node.path; - if (node.element->kind == ResolvedNodeKind::Scalar) { - for (uint i = 0; i < list.size(); ++i) { - if (std::optional scalar = scalar_value_to_double(list[i], node.element->scalar_kind); scalar.has_value()) { - RouteSeries *item_series = ensure_list_scalar_series(base_path, i, series); - if (node.element->scalar_kind == ScalarKind::Enum && !item_series->path.empty()) { - capture_enum_info(item_series->path, list[i], series); - } - append_fixed_scalar_point(item_series, tm, *scalar); - } - } - return; - } - for (uint i = 0; i < list.size(); ++i) { - const std::string item_path = base_path + "/" + std::to_string(i); - append_fast_node(*node.element, list[i], tm, series, &item_path); - } - return; - } - case ResolvedNodeKind::Ignore: - return; - } -} - -void append_event_fast_reader(cereal::Event::Which which, - const cereal::Event::Reader &event, - const SchemaIndex &schema, - const dbc::Database *can_dbc, - bool skip_raw_can, - double time_offset, - SeriesAccumulator *series) { - const uint16_t which_index = static_cast(which); - if (which_index >= schema.by_which.size() || !schema.by_which[which_index].has_value()) { - return; - } - const ResolvedService &service = *schema.by_which[which_index]; - const capnp::DynamicStruct::Reader dynamic_event(event); - const capnp::DynamicValue::Reader payload = dynamic_event.get(service.union_field); - const double tm = static_cast(event.getLogMonoTime()) / 1.0e9 - time_offset; - append_fixed_scalar_point(&series->fixed_series[static_cast(service.valid_slot)], - tm, - event.getValid() ? 1.0 : 0.0); - append_fixed_scalar_point(&series->fixed_series[static_cast(service.log_mono_time_slot)], - tm, - static_cast(event.getLogMonoTime())); - append_fixed_scalar_point(&series->fixed_series[static_cast(service.seconds_slot)], - tm, - tm); - if (service.service_name == "can" || service.service_name == "sendcan") { - const CanServiceKind can_service = service.service_name == "can" - ? CanServiceKind::Can - : CanServiceKind::Sendcan; - auto decode_message = [&](uint8_t bus, uint32_t address, const auto &dat_reader) { - const auto bytes = dat_reader.begin(); - decode_can_frame(can_dbc, service.service_name, bus, address, bytes, dat_reader.size(), tm, series); - }; - if (service.service_name == "can") { - for (const auto &msg : event.getCan()) { - append_can_frame(can_service, - static_cast(msg.getSrc()), - msg.getAddress(), - msg.getDeprecated().getBusTime(), - msg.getDat(), - tm, - series); - if (!skip_raw_can) continue; - decode_message(static_cast(msg.getSrc()), msg.getAddress(), msg.getDat()); - } - } else { - for (const auto &msg : event.getSendcan()) { - append_can_frame(can_service, - static_cast(msg.getSrc()), - msg.getAddress(), - msg.getDeprecated().getBusTime(), - msg.getDat(), - tm, - series); - if (!skip_raw_can) continue; - decode_message(static_cast(msg.getSrc()), msg.getAddress(), msg.getDat()); - } - } - if (skip_raw_can) { - return; - } - } - - append_fast_node(service.payload, payload, tm, series); -} - void append_event_fast(cereal::Event::Which which, int32_t eidx_segnum, kj::ArrayPtr data, - const SchemaIndex &schema, const dbc::Database *can_dbc, bool skip_raw_can, double time_offset, @@ -1353,14 +1034,13 @@ void append_event_fast(cereal::Event::Which which, return; } with_parseable_event(data, [&](const cereal::Event::Reader &event) { - append_event_fast_reader(which, event, schema, can_dbc, skip_raw_can, time_offset, series); + append_event_static_reader(which, event, can_dbc, skip_raw_can, time_offset, series); }); } void append_events_fast_range(const std::vector &events, size_t begin, size_t end, - const SchemaIndex &schema, const dbc::Database *can_dbc, bool skip_raw_can, SeriesAccumulator *series) { @@ -1369,7 +1049,6 @@ void append_events_fast_range(const std::vector &events, append_event_fast(event_record.which, event_record.eidx_segnum, event_record.data, - schema, can_dbc, skip_raw_can, 0.0, @@ -1802,7 +1481,7 @@ SeriesAccumulator extract_segment_series(const std::vector &events, 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(), schema, can_dbc, skip_raw_can, &series); + append_events_fast_range(events, 0, events.size(), can_dbc, skip_raw_can, &series); return series; } @@ -1819,10 +1498,10 @@ SeriesAccumulator extract_segment_series(const std::vector &events, 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, schema, can_dbc, skip_raw_can, &chunk_results[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), schema, can_dbc, skip_raw_can, &chunk_results[0]); + 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(); } @@ -2044,13 +1723,12 @@ void StreamAccumulator::appendEvent(kj::ArrayPtr data) { } } - append_event_fast_reader(which, - event, - impl_->schema, - impl_->can_dbc ? &*impl_->can_dbc : nullptr, - impl_->can_dbc.has_value(), - *impl_->time_offset, - &impl_->series); + 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();