support find_first_sensor_time

Author: berndpfrommer

include/event_camera_py/decoder.h

@@ -68,6 +68,24 @@ class Decoder
     return (std::tuple<bool, uint64_t>({reachedTimeLimit, nextTime}));
   }
 
+  std::tuple<bool, uint64_t> find_first_sensor_time(pybind11::object msg)
+  {
+    auto decoder = initialize_decoder(
+      get_attr<std::string>(msg, "encoding"), get_attr<uint32_t>(msg, "width"),
+      get_attr<uint32_t>(msg, "height"));
+
+    pybind11::object eventsObj = get_attr<pybind11::object>(msg, "events");
+    Py_buffer view;
+    if (PyObject_GetBuffer(eventsObj.ptr(), &view, PyBUF_CONTIG_RO) != 0) {
+      throw std::runtime_error("cannot convert events to byte buffer");
+    }
+    uint64_t firstTime{0};
+    const bool foundTime = decoder->findFirstSensorTime(
+      reinterpret_cast<const uint8_t *>(view.buf), view.len, &firstTime);
+    PyBuffer_Release(&view);
+    return (std::tuple<bool, uint64_t>({foundTime, firstTime}));
+  }
+
   void decode_bytes(
     const std::string & encoding, uint16_t width, uint16_t height, uint64_t timeBase,
     pybind11::bytes events)
@@ -125,6 +143,7 @@ class Decoder
   DecoderType * initialize_decoder(const std::string & encoding, uint32_t width, uint32_t height)
   {
     accumulator_.initialize(width, height);
+    // this will only create a decoder on the first call, subsequently return instance
     auto decoder = decoderFactory_.getInstance(encoding, width, height);
     if (!decoder) {
       throw(std::runtime_error("no decoder for encoding " + encoding));

src/decoder.cpp

@@ -94,6 +94,20 @@ void declare_decoder(pybind11::module & m, std::string typestr)
           time following time limit (only valid if time limit has been reached!)
         :rtype: tuple[boolean, uint64_t]
         )pbdoc")
+    .def("find_first_sensor_time", &MyDecoder::find_first_sensor_time, R"pbdoc(
+        find_first_sensor_time(msg) -> tuple[Boolean, uint64_t]
+
+        Peeks into encoded message to find first event sensor time stamp. The
+        boolean return flag indicates whether any time stamp was detected.
+        This function does not advance the state of the decoder, and will not
+        leverage state from previous message packets. It is typically used to cheaply find
+        the first time stamp in an EventPacket message without decoding the whole packet.
+
+        :param msg: event packet message to decode
+        :type msg:  event_camera_msgs/msgs/EventPacket
+        :return: tuple with flag (true if event sensor time was found) and the sensor time
+        :rtype: tuple[boolean, uint64_t]
+        )pbdoc")
     .def("get_start_time", &MyDecoder::get_start_time, R"pbdoc(
         get_start_time() -> uint64
         

tests/test_1.py

@@ -182,6 +182,20 @@ def test_unique_until(verbose=False):
     )
 
 
+def test_find_first_sensor_time(verbose=True):
+    bag = BagReader('tests/test_events_1', verbose)
+    if verbose:
+        print('Testing find_first_time_stamp')
+    decoder = Decoder()
+    for _, msg, _ in bag.read_messages(topics=['/event_camera/events']):
+        found_ts, ts = decoder.find_first_sensor_time(msg)
+        assert found_ts
+        if verbose:
+            print('first sensor time stamp: ', ts)
+        assert ts == 7139840
+        break
+
+
 if __name__ == '__main__':
     test_decode_bytes(True)
     test_decode_msg(True)