RSDK-7132 Stream Ticks

src/viam/sdk/components/board.hpp

@@ -3,6 +3,7 @@
 /// @brief Defines a `Board` component.
 #pragma once
 
+#include <chrono>
 #include <string>
 #include <unordered_map>
 
@@ -44,6 +45,19 @@ class Board : public Component {
         std::unordered_map<std::string, digital_value> digital_interrupt_values;
     };
 
+    /// @struct tick
+    /// @brief A board's digital interrupt.
+    struct Tick {
+        /// name of the digital interrupt pin.
+        std::string pin_name;
+
+        /// time in nanoseconds the tick occured. This does not represent an absolute time.
+        std::chrono::nanoseconds time;
+
+        /// bool high or low.
+        bool high;
+    };
+
     /// @enum power_mode
     /// @brief Power mode of the board
     /// The effect of these power modes depends on your physical board
@@ -221,6 +235,24 @@ class Board : public Component {
     virtual digital_value read_digital_interrupt(const std::string& digital_interrupt_name,
                                                  const AttributeMap& extra) = 0;
 
+    /// @brief Returns a stream of digital interrupt ticks.
+    /// @param digital_interrupt_names digital interrupts to stream
+    /// @param tick_handler callback function to call when a tick occurs.
+    /// This should return true to keep streaming ticks and false to indicate that the stream of
+    /// ticks should terminate. The callback function should not be blocking.
+    inline void stream_ticks(std::vector<std::string> const& digital_interrupt_names,
+                             std::function<bool(Tick&& tick)> const& tick_handler) {
+        return stream_ticks(digital_interrupt_names, tick_handler, {});
+    }
+
+    /// @brief Returns a stream of digital interrupt ticks.
+    /// @param digital_interrupt_names digital interrupts to stream
+    /// @param tick_handler callback function to call when a tick occurs.
+    /// @param extra Any additional arguments to the method
+    virtual void stream_ticks(std::vector<std::string> const& digital_interrupt_names,
+                              std::function<bool(Tick&& tick)> const& tick_handler,
+                              const AttributeMap& extra) = 0;
+
     /// @brief Sets the power consumption mode of the board to the requested setting for the given
     /// duration.
     /// @param power_mode Requested power mode

src/viam/sdk/components/private/board_client.cpp

@@ -137,6 +137,31 @@ Board::digital_value BoardClient::read_digital_interrupt(const std::string& digi
     return response.value();
 }
 
+void BoardClient::stream_ticks(std::vector<std::string> const& digital_interrupt_names,
+                               std::function<bool(Tick&& tick)> const& tick_handler,
+                               const AttributeMap& extra) {
+    viam::component::board::v1::StreamTicksRequest request;
+    viam::component::board::v1::StreamTicksResponse response;
+    ClientContext ctx;
+
+    request.set_name(this->name());
+
+    for (const auto& name : digital_interrupt_names) {
+        request.add_pin_names(name);
+    }
+    *request.mutable_extra() = map_to_struct(extra);
+
+    auto reader = stub_->StreamTicks(ctx, request);
+
+    while (reader->Read(&response)) {
+        if (!tick_handler({response.pin_name(),
+                           std::chrono::nanoseconds(response.time()),
+                           response.high()})) {
+            break;
+        }
+    };
+}
+
 void BoardClient::set_power_mode(power_mode power_mode,
                                  const AttributeMap& extra,
                                  const boost::optional<std::chrono::microseconds>& duration) {

src/viam/sdk/components/private/board_client.hpp

@@ -45,6 +45,10 @@ class BoardClient : public Board {
                         const boost::optional<std::chrono::microseconds>& duration) override;
     std::vector<GeometryConfig> get_geometries(const AttributeMap& extra) override;
 
+    void stream_ticks(std::vector<std::string> const& digital_interrupt_names,
+                      std::function<bool(Tick&& tick)> const& tick_handler,
+                      const AttributeMap& extra) override;
+
     // the `extra` param is frequently unnecessary but needs to be supported. Ideally, we'd
     // like to live in a world where implementers of derived classes don't need to go out of
     // their way to support two versions of a method (an `extra` version and a non-`extra`
@@ -65,6 +69,7 @@ class BoardClient : public Board {
     using Board::set_power_mode;
     using Board::set_pwm_duty_cycle;
     using Board::set_pwm_frequency;
+    using Board::stream_ticks;
     using Board::write_analog;
 
    private:

src/viam/sdk/components/private/board_server.cpp

@@ -174,6 +174,32 @@ ::grpc::Status BoardServer::GetDigitalInterruptValue(
     return ::grpc::Status();
 }
 
+::grpc::Status BoardServer::StreamTicks(
+    ::grpc::ServerContext* context,
+    const ::viam::component::board::v1::StreamTicksRequest* request,
+    ::grpc::ServerWriter<::viam::component::board::v1::StreamTicksResponse>* writer) noexcept {
+    make_service_helper<Board>(
+        "BoardServer::StreamTicks", this, request)([&](auto& helper, auto& board) {
+        const std::vector<std::string> digital_interrupt_names(request->pin_names().begin(),
+                                                               request->pin_names().end());
+        auto writeTick = [writer, context](Board::Tick&& tick) {
+            if (context->IsCancelled()) {
+                // send bool to tell the board to stop calling the callback function.
+                return false;
+            }
+            ::viam::component::board::v1::StreamTicksResponse response;
+            response.set_pin_name(std::move(tick.pin_name));
+            response.set_high(std::move(tick.high));
+            response.set_time(std::move(tick.time.count()));
+            writer->Write(response);
+            return true;
+        };
+        board->stream_ticks(digital_interrupt_names, writeTick, helper.getExtra());
+    });
+
+    return ::grpc::Status();
+}
+
 ::grpc::Status BoardServer::SetPowerMode(
     ::grpc::ServerContext*,
     const ::viam::component::board::v1::SetPowerModeRequest* request,

src/viam/sdk/components/private/board_server.hpp

@@ -74,6 +74,12 @@ class BoardServer : public ResourceServer,
         const ::viam::component::board::v1::GetDigitalInterruptValueRequest* request,
         ::viam::component::board::v1::GetDigitalInterruptValueResponse* response) override;
 
+    ::grpc::Status StreamTicks(
+        ::grpc::ServerContext* context,
+        const ::viam::component::board::v1::StreamTicksRequest* request,
+        ::grpc::ServerWriter<::viam::component::board::v1::StreamTicksResponse>* writer) noexcept
+        override;
+
     ::grpc::Status SetPowerMode(
         ::grpc::ServerContext* context,
         const ::viam::component::board::v1::SetPowerModeRequest* request,

src/viam/sdk/tests/mocks/mock_board.cpp

@@ -71,6 +71,14 @@ Board::digital_value MockBoard::read_digital_interrupt(const std::string& digita
     return this->peek_read_digital_interrupt_ret;
 }
 
+void MockBoard::stream_ticks(std::vector<std::string> const& digital_interrupt_names,
+                             std::function<bool(Tick&& tick)> const& tick_handler,
+                             const AttributeMap& extra) {
+    for (const auto& name : digital_interrupt_names) {
+        peek_callbacks[name] = tick_handler;
+    }
+}
+
 void MockBoard::set_power_mode(power_mode power_mode,
                                const AttributeMap&,
                                const boost::optional<std::chrono::microseconds>& duration) {

src/viam/sdk/tests/mocks/mock_board.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <map>
 #include <viam/sdk/common/proto_type.hpp>
 #include <viam/sdk/components/board.hpp>
 
@@ -27,13 +28,17 @@ class MockBoard : public viam::sdk::Board {
     void write_analog(const std::string& pin, int value, const sdk::AttributeMap& extra) override;
     Board::digital_value read_digital_interrupt(const std::string& digital_interrupt_name,
                                                 const sdk::AttributeMap& extra) override;
+    void stream_ticks(std::vector<std::string> const& digital_interrupt_names,
+                      std::function<bool(Board::Tick&& tick)> const& tick_handler,
+                      const sdk::AttributeMap& extra) override;
     void set_power_mode(power_mode power_mode,
                         const sdk::AttributeMap& extra,
                         const boost::optional<std::chrono::microseconds>& duration) override;
     std::vector<sdk::GeometryConfig> get_geometries(const sdk::AttributeMap& extra) override;
 
     std::string peek_pin, peek_analog_reader_name, peek_digital_interrupt_name;
     int peek_pin_value;
+    std::map<std::string, std::function<bool(Board::Tick tick)>> peek_callbacks;
     Board::status peek_get_status_ret;
     bool peek_set_gpio_high;
     bool peek_get_gpio_ret;

src/viam/sdk/tests/test_board.cpp

@@ -146,6 +146,22 @@ BOOST_AUTO_TEST_CASE(test_read_digital_interrupt) {
     });
 }
 
+BOOST_AUTO_TEST_CASE(test_stream_ticks) {
+    const auto mock = std::make_shared<MockBoard>("mock_board");
+
+    client_to_mock_pipeline<Board>(mock, [&](Board& client) {
+        auto tick_handler = [](Board::Tick tick) -> bool { return false; };
+
+        std::vector<std::string> pin_names = {"t1", "t2"};
+        mock->sdk::Board::stream_ticks(pin_names, tick_handler);
+
+        auto iterator = mock->peek_callbacks.begin();
+        BOOST_CHECK_EQUAL(iterator->first, "t1");
+        iterator++;
+        BOOST_CHECK_EQUAL(iterator->first, "t2");
+    });
+}
+
 BOOST_AUTO_TEST_CASE(test_get_analog_reader_names) {
     const auto mock = std::make_shared<MockBoard>("mock_board");
     client_to_mock_pipeline<Board>(mock, [&](Board& client) {