feat: add new operators

Author: rcarcasses

libs/api/include/rtbot/OperatorJson.h

@@ -33,6 +33,7 @@
 #include "rtbot/std/MovingAverage.h"
 #include "rtbot/std/MovingSum.h"
 #include "rtbot/std/PeakDetector.h"
+#include "rtbot/std/PctChange.h"
 #include "rtbot/std/Replace.h"
 #include "rtbot/std/ResamplerConstant.h"
 #include "rtbot/std/ResamplerHermite.h"
@@ -186,6 +187,8 @@ class OperatorJson {
       return make_resampler_constant(id, parsed["interval"].get<int>());
     } else if (type == "ResamplerHermite") {
       return make_resampler_hermite(id, parsed["interval"].get<int>());
+    } else if (type == "PctChange") {
+      return make_pct_change(id);
     } else if (type == "Pipeline") {
       // Validate port types
       auto input_types = parsed["input_port_types"].get<std::vector<std::string>>();
@@ -337,6 +340,8 @@ class OperatorJson {
       }
     } else if (type == "ResamplerHermite") {
       j["interval"] = std::dynamic_pointer_cast<ResamplerHermite>(op)->get_interval();
+    } else if (type == "PctChange") {
+      // PctChange has no additional parameters to serialize
     } else if (type == "Pipeline") {
       auto pipeline = std::dynamic_pointer_cast<Pipeline>(op);
       j["type"] = "Pipeline";

libs/api/test/test_operator_json.cpp

@@ -0,0 +1,21 @@
+#include <catch2/catch.hpp>
+
+#include "rtbot/OperatorJson.h"
+
+using namespace rtbot;
+
+TEST_CASE("OperatorJson creates PctChange operator", "[OperatorJson][PctChange]") {
+  const std::string json = R"({
+    "type": "PctChange",
+    "id": "pct_op"
+  })";
+
+  auto op = OperatorJson::read_op(json);
+  REQUIRE(op->type_name() == "PctChange");
+  REQUIRE(op->id() == "pct_op");
+
+  const auto serialized = OperatorJson::write_op(op);
+  auto round_trip = OperatorJson::read_op(serialized);
+  REQUIRE(round_trip->type_name() == "PctChange");
+  REQUIRE(round_trip->id() == "pct_op");
+}

libs/std/include/rtbot/std/Clip.h

@@ -0,0 +1,104 @@
+#ifndef CLIP_H
+#define CLIP_H
+
+#include <algorithm>
+#include <optional>
+
+#include "rtbot/Message.h"
+#include "rtbot/Operator.h"
+#include "rtbot/PortType.h"
+
+namespace rtbot {
+
+class Clip : public Operator {
+ public:
+  Clip(std::string id, std::optional<double> lower = std::nullopt, std::optional<double> upper = std::nullopt)
+      : Operator(std::move(id)), lower_(lower), upper_(upper) {
+    add_data_port<NumberData>();
+    add_output_port<NumberData>();
+  }
+
+  void reset() override { Operator::reset(); }
+
+  std::string type_name() const override { return "Clip"; }
+
+  std::optional<double> lower() const { return lower_; }
+  std::optional<double> upper() const { return upper_; }
+
+  Bytes collect() override {
+    Bytes bytes = Operator::collect();
+
+    bool has_lower = lower_.has_value();
+    bool has_upper = upper_.has_value();
+    bytes.push_back(static_cast<uint8_t>(has_lower));
+    if (has_lower) {
+      bytes.insert(bytes.end(), reinterpret_cast<const uint8_t*>(&(*lower_)),
+                   reinterpret_cast<const uint8_t*>(&(*lower_)) + sizeof(double));
+    }
+
+    bytes.push_back(static_cast<uint8_t>(has_upper));
+    if (has_upper) {
+      bytes.insert(bytes.end(), reinterpret_cast<const uint8_t*>(&(*upper_)),
+                   reinterpret_cast<const uint8_t*>(&(*upper_)) + sizeof(double));
+    }
+
+    return bytes;
+  }
+
+  void restore(Bytes::const_iterator& it) override {
+    Operator::restore(it);
+
+    bool has_lower = static_cast<bool>(*it++);
+    if (has_lower) {
+      lower_ = *reinterpret_cast<const double*>(&(*it));
+      it += sizeof(double);
+    } else {
+      lower_.reset();
+    }
+
+    bool has_upper = static_cast<bool>(*it++);
+    if (has_upper) {
+      upper_ = *reinterpret_cast<const double*>(&(*it));
+      it += sizeof(double);
+    } else {
+      upper_.reset();
+    }
+  }
+
+ protected:
+  void process_data() override {
+    auto& input_queue = get_data_queue(0);
+    auto& output_queue = get_output_queue(0);
+
+    while (!input_queue.empty()) {
+      const auto* msg = dynamic_cast<const Message<NumberData>*>(input_queue.front().get());
+      if (!msg) {
+        throw std::runtime_error("Invalid message type in Clip operator.");
+      }
+
+      double value = msg->data.value;
+      if (lower_) {
+        value = std::max(value, *lower_);
+      }
+      if (upper_) {
+        value = std::min(value, *upper_);
+      }
+
+      output_queue.push_back(create_message<NumberData>(msg->time, NumberData{value}));
+      input_queue.pop_front();
+    }
+  }
+
+ private:
+  std::optional<double> lower_;
+  std::optional<double> upper_;
+};
+
+inline std::shared_ptr<Operator> make_clip(const std::string& id, std::optional<double> lower = std::nullopt,
+                                          std::optional<double> upper = std::nullopt) {
+  return std::make_shared<Clip>(id, lower, upper);
+}
+
+}  // namespace rtbot
+
+#endif  // CLIP_H

libs/std/include/rtbot/std/Clip.md

@@ -0,0 +1,15 @@
+# Clip Operator
+
+The `Clip` operator constrains numeric values to a closed interval. Both lower and upper bounds are optional.
+
+## Parameters
+- `lower` (number, optional): Minimum value. If omitted the input is not bounded below.
+- `upper` (number, optional): Maximum value. If omitted the input is not bounded above.
+
+## Ports
+- Input `NumberData`
+- Output `NumberData`
+
+## Behaviour
+- Each incoming sample is clamped between `lower` and `upper` if the bounds are provided.
+- Bounds may be given independently (only lower or only upper).

libs/std/include/rtbot/std/CumulativeProduct.h

@@ -0,0 +1,67 @@
+#ifndef CUMULATIVE_PRODUCT_H
+#define CUMULATIVE_PRODUCT_H
+
+#include "rtbot/Message.h"
+#include "rtbot/Operator.h"
+#include "rtbot/PortType.h"
+
+namespace rtbot {
+
+class CumulativeProduct : public Operator {
+ public:
+  explicit CumulativeProduct(std::string id) : Operator(std::move(id)), product_(1.0) {
+    add_data_port<NumberData>();
+    add_output_port<NumberData>();
+  }
+
+  void reset() override {
+    Operator::reset();
+    product_ = 1.0;
+  }
+
+  std::string type_name() const override { return "CumulativeProduct"; }
+
+  double current_product() const { return product_; }
+
+  Bytes collect() override {
+    Bytes bytes = Operator::collect();
+    bytes.insert(bytes.end(), reinterpret_cast<const uint8_t*>(&product_),
+                 reinterpret_cast<const uint8_t*>(&product_) + sizeof(product_));
+    return bytes;
+  }
+
+  void restore(Bytes::const_iterator& it) override {
+    Operator::restore(it);
+    product_ = *reinterpret_cast<const double*>(&(*it));
+    it += sizeof(double);
+  }
+
+ protected:
+  void process_data() override {
+    auto& input_queue = get_data_queue(0);
+    auto& output_queue = get_output_queue(0);
+
+    while (!input_queue.empty()) {
+      const auto* msg = dynamic_cast<const Message<NumberData>*>(input_queue.front().get());
+      if (!msg) {
+        throw std::runtime_error("Invalid message type in CumulativeProduct");
+      }
+
+      product_ *= msg->data.value;
+      output_queue.push_back(create_message<NumberData>(msg->time, NumberData{product_}));
+
+      input_queue.pop_front();
+    }
+  }
+
+ private:
+  double product_;
+};
+
+inline std::shared_ptr<Operator> make_cumulative_product(const std::string& id) {
+  return std::make_shared<CumulativeProduct>(id);
+}
+
+}  // namespace rtbot
+
+#endif  // CUMULATIVE_PRODUCT_H

libs/std/include/rtbot/std/CumulativeProduct.md

@@ -0,0 +1,11 @@
+# CumulativeProduct Operator
+
+`CumulativeProduct` maintains the running product of all received numeric samples.
+
+## Ports
+- Input `NumberData`
+- Output `NumberData`
+
+## Behaviour
+- Multiplies the incoming value into the internal state and emits the updated product for each message.
+- The running product resets to `1.0` when the operator is reset.

libs/std/include/rtbot/std/MovingMax.h

@@ -0,0 +1,48 @@
+#ifndef MOVING_MAX_H
+#define MOVING_MAX_H
+
+#include <algorithm>
+#include <vector>
+
+#include "rtbot/Buffer.h"
+#include "rtbot/Message.h"
+#include "rtbot/Operator.h"
+
+namespace rtbot {
+
+struct MovingMaxFeatures {
+  static constexpr bool TRACK_SUM = false;
+  static constexpr bool TRACK_VARIANCE = false;
+};
+
+class MovingMax : public Buffer<NumberData, MovingMaxFeatures> {
+ public:
+  MovingMax(std::string id, size_t window_size)
+      : Buffer<NumberData, MovingMaxFeatures>(std::move(id), window_size) {}
+
+  std::string type_name() const override { return "MovingMax"; }
+
+ protected:
+  std::vector<std::unique_ptr<Message<NumberData>>> process_message(const Message<NumberData>* msg) override {
+    if (!this->buffer_full()) {
+      return {};
+    }
+
+    double max_value = this->buffer().front()->data.value;
+    for (const auto& entry : this->buffer()) {
+      max_value = std::max(max_value, entry->data.value);
+    }
+
+    std::vector<std::unique_ptr<Message<NumberData>>> output;
+    output.push_back(create_message<NumberData>(msg->time, NumberData{max_value}));
+    return output;
+  }
+};
+
+inline std::shared_ptr<Operator> make_moving_max(const std::string& id, size_t window_size) {
+  return std::make_shared<MovingMax>(id, window_size);
+}
+
+}  // namespace rtbot
+
+#endif  // MOVING_MAX_H

libs/std/include/rtbot/std/MovingMax.md

@@ -0,0 +1,15 @@
+# MovingMax Operator
+
+The `MovingMax` operator computes the maximum of the latest `window_size` samples. It behaves similarly to a pandas
+rolling maximum with `min_periods=window_size`.
+
+## Parameters
+- `window_size` (integer, required): Number of samples in the sliding window.
+
+## Ports
+- Input `NumberData`
+- Output `NumberData`
+
+## Behaviour
+- Emits no data until the buffer contains `window_size` samples.
+- Afterwards emits the maximum of the current window for each new input sample.

libs/std/include/rtbot/std/MovingMin.h

@@ -0,0 +1,48 @@
+#ifndef MOVING_MIN_H
+#define MOVING_MIN_H
+
+#include <algorithm>
+#include <vector>
+
+#include "rtbot/Buffer.h"
+#include "rtbot/Message.h"
+#include "rtbot/Operator.h"
+
+namespace rtbot {
+
+struct MovingMinFeatures {
+  static constexpr bool TRACK_SUM = false;
+  static constexpr bool TRACK_VARIANCE = false;
+};
+
+class MovingMin : public Buffer<NumberData, MovingMinFeatures> {
+ public:
+  MovingMin(std::string id, size_t window_size)
+      : Buffer<NumberData, MovingMinFeatures>(std::move(id), window_size) {}
+
+  std::string type_name() const override { return "MovingMin"; }
+
+ protected:
+  std::vector<std::unique_ptr<Message<NumberData>>> process_message(const Message<NumberData>* msg) override {
+    if (!this->buffer_full()) {
+      return {};
+    }
+
+    double min_value = this->buffer().front()->data.value;
+    for (const auto& entry : this->buffer()) {
+      min_value = std::min(min_value, entry->data.value);
+    }
+
+    std::vector<std::unique_ptr<Message<NumberData>>> output;
+    output.push_back(create_message<NumberData>(msg->time, NumberData{min_value}));
+    return output;
+  }
+};
+
+inline std::shared_ptr<Operator> make_moving_min(const std::string& id, size_t window_size) {
+  return std::make_shared<MovingMin>(id, window_size);
+}
+
+}  // namespace rtbot
+
+#endif  // MOVING_MIN_H

libs/std/include/rtbot/std/MovingMin.md

@@ -0,0 +1,15 @@
+# MovingMin Operator
+
+The `MovingMin` operator computes the minimum of the most recent `window_size` values. It emits a value once the
+internal buffer is full and thereafter slides forward one sample at a time.
+
+## Parameters
+- `window_size` (integer, required): Number of samples inside the moving window.
+
+## Ports
+- Input `NumberData`
+- Output `NumberData`
+
+## Behaviour
+- Produces no output until `window_size` samples have been observed.
+- After the buffer is full, outputs the minimum of the current window for each new sample.