jp: 2x faster parsing (#37904)

* jp: 2x faster parsing

* rm dynamic path

* cleanup

* lil more

* livin in the future

* clean that up

* one more

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/

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)

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<double>(static_cast<uint16_t>({value_expr}))"
+    return f"static_cast<double>({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<decltype({reader_expr}.which())>({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<uint8_t>(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<uint8_t>(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<cereal::Event::Which>({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<double>(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<std::string> &static_event_fixed_paths() {")
+    self.emit(2, "static const std::vector<std::string> 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<std::string_view> 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<double>(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]} <repo-root> <output>", 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)

tools/jotpluggler/sketch_layout.cc

@@ -2,7 +2,6 @@
 #include "tools/jotpluggler/car_fingerprint_to_dbc.h"
 #include "tools/jotpluggler/common.h"
 
-#include <capnp/dynamic.h>
 #include <kj/exception.h>
 
 #include <chrono>
@@ -10,6 +9,7 @@
 #include <cstdlib>
 #include <cstring>
 #include <iostream>
+#include <initializer_list>
 #include <map>
 #include <mutex>
 #include <limits>
@@ -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<ResolvedNode> children;
-  std::unique_ptr<ResolvedNode> 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<std::optional<ResolvedService>> by_which;
   size_t fixed_series_count = 0;
   std::vector<std::string> fixed_paths;
 
@@ -112,6 +72,27 @@ struct SeriesAccumulator {
   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;
@@ -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<std::string> *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<int>((*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<ResolvedNode>(
-      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<std::string> *fixed_paths) {
-  const int slot = static_cast<int>((*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<cereal::Event>().asStruct();
-    uint16_t max_discriminant = 0;
-    for (auto union_field : event_schema.getUnionFields()) {
-      max_discriminant = std::max<uint16_t>(max_discriminant, union_field.getProto().getDiscriminantValue());
-    }
-    out.by_which.resize(static_cast<size_t>(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<double> scalar_value_to_double(const capnp::DynamicValue::Reader &value, ScalarKind kind) {
-  switch (kind) {
-    case ScalarKind::Bool:
-      return value.as<bool>() ? 1.0 : 0.0;
-    case ScalarKind::Int:
-      return static_cast<double>(value.as<int64_t>());
-    case ScalarKind::UInt:
-      return static_cast<double>(value.as<uint64_t>());
-    case ScalarKind::Float:
-      return value.as<double>();
-    case ScalarKind::Enum:
-      return static_cast<double>(value.as<capnp::DynamicEnum>().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<capnp::DynamicEnum>();
-  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<size_t>(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<size_t>(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<double> 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<capnp::DynamicStruct>();
-      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<capnp::DynamicList>();
-      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<double> 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<uint16_t>(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<double>(event.getLogMonoTime()) / 1.0e9 - time_offset;
-  append_fixed_scalar_point(&series->fixed_series[static_cast<size_t>(service.valid_slot)],
-                            tm,
-                            event.getValid() ? 1.0 : 0.0);
-  append_fixed_scalar_point(&series->fixed_series[static_cast<size_t>(service.log_mono_time_slot)],
-                            tm,
-                            static_cast<double>(event.getLogMonoTime()));
-  append_fixed_scalar_point(&series->fixed_series[static_cast<size_t>(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<uint8_t>(msg.getSrc()),
-                         msg.getAddress(),
-                         msg.getDeprecated().getBusTime(),
-                         msg.getDat(),
-                         tm,
-                         series);
-        if (!skip_raw_can) continue;
-        decode_message(static_cast<uint8_t>(msg.getSrc()), msg.getAddress(), msg.getDat());
-      }
-    } else {
-      for (const auto &msg : event.getSendcan()) {
-        append_can_frame(can_service,
-                         static_cast<uint8_t>(msg.getSrc()),
-                         msg.getAddress(),
-                         msg.getDeprecated().getBusTime(),
-                         msg.getDat(),
-                         tm,
-                         series);
-        if (!skip_raw_can) continue;
-        decode_message(static_cast<uint8_t>(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<const capnp::word> 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<Event> &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<Event> &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<Event> &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<Event> &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<const capnp::word> 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();