CANParser: optimize data handling by eliminating data copy after Update()

opendbc/can/common.h

@@ -40,15 +40,20 @@ struct CanData {
   std::vector<CanFrame> frames;
 };
 
+struct SignalValue {
+  uint64_t ts_nanos;
+  double value;
+  std::vector<double> all_values;
+};
+
 class MessageState {
 public:
   std::string name;
   uint32_t address;
   unsigned int size;
 
   std::vector<Signal> parse_sigs;
-  std::vector<double> vals;
-  std::vector<std::vector<double>> all_vals;
+  std::unordered_map<std::string, SignalValue> signal_values;
 
   uint64_t last_seen_nanos;
   uint64_t check_threshold;

opendbc/can/common.pxd

@@ -62,11 +62,14 @@ cdef extern from "common_dbc.h":
 cdef extern from "common.h":
   cdef const DBC* dbc_lookup(const string) except +
 
+  cdef struct SignalValue:
+    uint64_t ts_nanos
+    double value
+    vector[double] all_values
+
   cdef cppclass MessageState:
     vector[Signal] parse_sigs
-    vector[double] vals
-    vector[vector[double]] all_vals
-    uint64_t last_seen_nanos
+    unordered_map[string, SignalValue] signal_values
 
   cdef struct CanFrame:
     long src

opendbc/can/parser.cc

@@ -64,8 +64,10 @@ bool MessageState::parse(uint64_t nanos, const std::vector<uint8_t> &dat) {
   }
 
   for (int i = 0; i < parse_sigs.size(); i++) {
-    vals[i] = tmp_vals[i];
-    all_vals[i].push_back(vals[i]);
+    auto &val = signal_values.at(parse_sigs[i].name);
+    val.ts_nanos = nanos;
+    val.value = tmp_vals[i];
+    val.all_values.push_back(val.value);
   }
   last_seen_nanos = nanos;
 
@@ -121,8 +123,9 @@ CANParser::CANParser(int abus, const std::string& dbc_name, const std::vector<st
 
     // track all signals for this message
     state.parse_sigs = msg->sigs;
-    state.vals.resize(msg->sigs.size());
-    state.all_vals.resize(msg->sigs.size());
+    for (auto &sig : msg->sigs) {
+      state.signal_values[sig.name] = {};
+    }
   }
 }
 
@@ -142,10 +145,9 @@ CANParser::CANParser(int abus, const std::string& dbc_name, bool ignore_checksum
       .ignore_counter = ignore_counter,
     };
 
-    for (const auto& sig : msg.sigs) {
-      state.parse_sigs.push_back(sig);
-      state.vals.push_back(0);
-      state.all_vals.push_back({});
+    state.parse_sigs = msg.sigs;
+    for (auto &sig : msg.sigs) {
+      message_states[state.address].signal_values[sig.name] = {};
     }
 
     message_states[state.address] = state;
@@ -155,7 +157,9 @@ CANParser::CANParser(int abus, const std::string& dbc_name, bool ignore_checksum
 std::set<uint32_t> CANParser::update(const std::vector<CanData> &can_data) {
   // Clear all_values
   for (auto &state : message_states) {
-    for (auto &vals : state.second.all_vals) vals.clear();
+    for (auto &value : state.second.signal_values) {
+      value.second.all_values.clear();
+    }
   }
 
   std::set<uint32_t> updated_addresses;

opendbc/can/parser_pyx.pyx

@@ -6,12 +6,53 @@ from libcpp.string cimport string
 from libcpp.vector cimport vector
 from libc.stdint cimport uint32_t
 
-from .common cimport CANParser as cpp_CANParser
-from .common cimport dbc_lookup, Msg, DBC, CanData
+from .common cimport CANParser as cpp_CANParser, MessageState as cpp_MessageState
+from .common cimport dbc_lookup, DBC, CanData
 
 import numbers
+from collections.abc import Mapping
 from collections import defaultdict
 
+cdef class MessageState:
+  cdef cpp_MessageState *_state
+  cdef list _signal_names
+
+  def value(self, key):
+    return self._state.signal_values.at(key).value
+
+  def ts_nanos(self, key):
+    return self._state.signal_values.at(key).ts_nanos
+
+  def all_values(self, key):
+    return self._state.signal_values.at(key).all_values
+
+  @property
+  def signal_names(self):
+    return self._signal_names
+
+  @staticmethod
+  cdef create(cpp_MessageState *state):
+    message_state = MessageState()
+    message_state._state = state
+    message_state._signal_names = [it.first.decode("utf-8") for it in state.signal_values]
+    return message_state
+
+
+class ReadonlyDict(Mapping):
+  def __init__(self, state, value_accessor):
+    self._state = state
+    self._keys = state.signal_names
+    self._get_value = value_accessor
+
+  def __getitem__(self, key):
+    return self._get_value(self._state, key)
+
+  def __iter__(self):
+    return iter(self._keys)
+
+  def __len__(self):
+    return len(self._keys)
+
 
 cdef class CANParser:
   cdef:
@@ -47,21 +88,19 @@ cdef class CANParser:
       except IndexError:
         raise RuntimeError(f"could not find message {repr(c[0])} in DBC {self.dbc_name}")
 
-      address = m.address
-      message_v.push_back((address, c[1]))
-      self.addresses.add(address)
+      message_v.push_back((m.address, c[1]))
+      self.addresses.add(m.address)
 
-      name = m.name.decode("utf8")
-      signal_names = [sig.name.decode("utf-8") for sig in (<Msg*>m).sigs]
+    self.can = new cpp_CANParser(bus, dbc_name, message_v)
 
-      self.vl[address] = {name: 0.0 for name in signal_names}
-      self.vl[name] = self.vl[address]
-      self.vl_all[address] = defaultdict(list)
-      self.vl_all[name] = self.vl_all[address]
-      self.ts_nanos[address] = {name: 0.0 for name in signal_names}
-      self.ts_nanos[name] = self.ts_nanos[address]
+    for addr in self.addresses:
+      msg = self.dbc.addr_to_msg.at(addr)
+      msg_name = msg.name.decode("utf8")
+      message_state = MessageState.create(self.can.getMessageState(addr))
 
-    self.can = new cpp_CANParser(bus, dbc_name, message_v)
+      self.vl[msg_name] = self.vl[addr] = ReadonlyDict(message_state, MessageState.value)
+      self.vl_all[msg_name] = self.vl_all[addr] = ReadonlyDict(message_state, MessageState.all_values)
+      self.ts_nanos[msg_name] = self.ts_nanos[addr] = ReadonlyDict(message_state, MessageState.ts_nanos)
 
   def __dealloc__(self):
     if self.can:
@@ -71,9 +110,6 @@ cdef class CANParser:
     # input format:
     # [nanos, [[address, data, src], ...]]
     # [[nanos, [[address, data, src], ...], ...]]
-    for address in self.addresses:
-      self.vl_all[address].clear()
-
     cdef vector[CanData] can_data_array
 
     try:
@@ -95,20 +131,7 @@ cdef class CANParser:
     except TypeError:
       raise RuntimeError("invalid parameter")
 
-    updated_addrs = self.can.update(can_data_array)
-    for addr in updated_addrs:
-      vl = self.vl[addr]
-      vl_all = self.vl_all[addr]
-      ts_nanos = self.ts_nanos[addr]
-
-      state = self.can.getMessageState(addr)
-      for i in range(state.parse_sigs.size()):
-        name = <unicode>state.parse_sigs[i].name
-        vl[name] = state.vals[i]
-        vl_all[name] = state.all_vals[i]
-        ts_nanos[name] = state.last_seen_nanos
-
-    return updated_addrs
+    return self.can.update(can_data_array)
 
   @property
   def can_valid(self):

opendbc/can/tests/test_checksums.py

@@ -17,7 +17,7 @@ def verify_checksum(self, subtests, dbc_file: str, msg_name: str, msg_addr: int,
     for data in test_messages:
       expected_msg = (msg_addr, data, 0)
       parser.update_strings([0, [expected_msg]])
-      expected = copy.deepcopy(parser.vl[msg_name])
+      expected = {key: parser.vl[msg_name][key] for key in parser.vl[msg_name]}
 
       modified = copy.deepcopy(expected)
       modified.pop(checksum_field, None)