feat(rtbot): operator == implementation

Author: eduarduseldorf

libs/api/include/rtbot/Program.h

@@ -114,14 +114,14 @@ class Program {
 
   // Message processing
   ProgramMsgBatch receive(const Message<NumberData>& msg, const std::string& port_id = "i1") {
-    send_to_entry(msg, port_id);
+    send_to_entry(msg, port_id, false);
     ProgramMsgBatch result = collect_outputs(false);
     clear_all_outputs();
     return result;
   }
 
   ProgramMsgBatch receive_debug(const Message<NumberData>& msg, const std::string& port_id = "i1") {
-    send_to_entry(msg, port_id);
+    send_to_entry(msg, port_id, true);
     ProgramMsgBatch result = collect_outputs(true);
     clear_all_outputs();
     return result;
@@ -229,10 +229,10 @@ class Program {
     throw runtime_error("Could not resolve operator ID: " + id);
   }
 
-  void send_to_entry(const Message<NumberData>& msg, const std::string& port_id) {
+  void send_to_entry(const Message<NumberData>& msg, const std::string& port_id, bool debug=false) {
     auto port_info = OperatorJson::parse_port_name(port_id);
     operators_[entry_operator_id_]->receive_data(create_message<NumberData>(msg.time, msg.data), port_info.index);
-    operators_[entry_operator_id_]->execute();
+    operators_[entry_operator_id_]->execute(debug);
   }
 
   ProgramMsgBatch collect_outputs(bool debug_mode = false) {
@@ -255,7 +255,7 @@ class Program {
               // In debug mode, collect all ports
               if (debug_mode) {
                 for (size_t i = 0; i < op->num_output_ports(); i++) {
-                  const auto& queue = op->get_output_queue(i);
+                  const auto& queue = op->get_debug_output_queue(i);
                   if (!queue.empty()) {
                     PortMsgBatch port_msgs;
                     for (const auto& msg : queue) {

libs/core/include/rtbot/Buffer.h

@@ -72,11 +72,38 @@ class Buffer : public Operator {
     return std::sqrt(variance());
   }
 
-  Bytes collect() override {
-    Bytes bytes = Operator::collect();
+  bool equals(const Buffer& other) const {
+      
+    if (window_size_ != other.window_size_) return false;
+
+    if (buffer_.size() != other.buffer_.size()) return false;
+
+    auto it1 = buffer_.begin();
+    auto it2 = other.buffer_.begin();
+
+    for (; it1 != buffer_.end() && it2 != other.buffer_.end(); ++it1, ++it2) {
+        const auto& msg1 = *it1;
+        const auto& msg2 = *it2;
 
-    bytes.insert(bytes.end(), reinterpret_cast<const uint8_t*>(&window_size_),
-                 reinterpret_cast<const uint8_t*>(&window_size_) + sizeof(window_size_));
+        if (msg1 && msg2) {
+            if (msg1->time != msg2->time) return false;
+            if (msg1->hash() != msg2->hash()) return false;
+        } else return false;
+    }
+    
+    if constexpr (Features::TRACK_SUM) {
+        if (StateSerializer::hash_double(sum_) != StateSerializer::hash_double(other.sum_)) return false;
+    }
+
+    if constexpr (Features::TRACK_VARIANCE) {
+        if (StateSerializer::hash_double(M2_) != StateSerializer::hash_double(other.M2_)) return false;    
+    }
+
+    return Operator::equals(other);
+  }
+
+  Bytes collect() override {
+    Bytes bytes = Operator::collect();    
 
     size_t buffer_size = buffer_.size();
     bytes.insert(bytes.end(), reinterpret_cast<const uint8_t*>(&buffer_size),
@@ -90,65 +117,68 @@ class Buffer : public Operator {
       bytes.insert(bytes.end(), msg_bytes.begin(), msg_bytes.end());
     }
 
-    if constexpr (Features::TRACK_SUM) {
-      bytes.insert(bytes.end(), reinterpret_cast<const uint8_t*>(&sum_),
-                   reinterpret_cast<const uint8_t*>(&sum_) + sizeof(sum_));
-    }
-
-    if constexpr (Features::TRACK_VARIANCE) {
-      bytes.insert(bytes.end(), reinterpret_cast<const uint8_t*>(&M2_),
-                   reinterpret_cast<const uint8_t*>(&M2_) + sizeof(M2_));
-    }
-
     return bytes;
   }
 
   void restore(Bytes::const_iterator& it) override {
-    Operator::restore(it);
-
-    window_size_ = *reinterpret_cast<const size_t*>(&(*it));
-    it += sizeof(size_t);
+    // Call base restore first
+    Operator::restore(it);    
 
-    size_t buffer_size = *reinterpret_cast<const size_t*>(&(*it));
-    it += sizeof(size_t);
+    // ---- Read buffer_size safely ----
+    size_t buffer_size;
+    std::memcpy(&buffer_size, &(*it), sizeof(buffer_size));
+    it += sizeof(buffer_size);
 
+    // ---- Deserialize buffer ----
     buffer_.clear();
     for (size_t i = 0; i < buffer_size; ++i) {
-      size_t msg_size = *reinterpret_cast<const size_t*>(&(*it));
-      it += sizeof(size_t);
-
-      Bytes msg_bytes(it, it + msg_size);
-      buffer_.push_back(
-          std::unique_ptr<Message<T>>(dynamic_cast<Message<T>*>(BaseMessage::deserialize(msg_bytes).release())));
-      it += msg_size;
+        // Read size of each message
+        size_t msg_size;
+        std::memcpy(&msg_size, &(*it), sizeof(msg_size));
+        it += sizeof(msg_size);
+
+        // Extract message bytes
+        Bytes msg_bytes(it, it + msg_size);
+
+        // Deserialize message and cast to derived type
+        buffer_.push_back(
+            std::unique_ptr<Message<T>>(
+                dynamic_cast<Message<T>*>(BaseMessage::deserialize(msg_bytes).release())
+            )
+        );
+
+        it += msg_size;
     }
 
+    // ---- Optional statistics ----
     if constexpr (Features::TRACK_SUM) {
-      sum_ = *reinterpret_cast<const double*>(&(*it));
-      it += sizeof(double);
+        sum_ = 0.0;
+        if (!buffer_.empty()) {
+            // First pass: compute sum
+            for (const auto& msg : buffer_) {
+                sum_ += msg->data.value;
+            }
+        }
     }
 
     if constexpr (Features::TRACK_VARIANCE) {
-      M2_ = *reinterpret_cast<const double*>(&(*it));
-      it += sizeof(double);
-
-      // Recompute statistics from buffer to ensure consistency
-      sum_ = 0.0;
-      M2_ = 0.0;
-
-      if (!buffer_.empty()) {
-        // First pass: compute mean
-        for (const auto& msg : buffer_) {
-          sum_ += msg->data.value;
+        // Recompute statistics from buffer to ensure consistency
+        sum_ = 0.0;
+        M2_ = 0.0;
+
+        if (!buffer_.empty()) {
+            // First pass: compute sum
+            for (const auto& msg : buffer_) {
+                sum_ += msg->data.value;
+            }
+
+            // Second pass: compute M2
+            double mean = sum_ / buffer_.size();
+            for (const auto& msg : buffer_) {
+                double delta = msg->data.value - mean;
+                M2_ += delta * delta;
+            }
         }
-
-        // Second pass: compute M2
-        double mean = sum_ / buffer_.size();
-        for (const auto& msg : buffer_) {
-          double delta = msg->data.value - mean;
-          M2_ += delta * delta;
-        }
-      }
     }
   }
 

libs/core/include/rtbot/Demultiplexer.h

@@ -35,28 +35,18 @@ class Demultiplexer : public Operator {
 
   std::string type_name() const override { return "Demultiplexer"; }
 
-  size_t get_num_ports() const { return num_control_ports(); }  
+  size_t get_num_ports() const { return num_control_ports(); }
 
-  void receive_control(std::unique_ptr<BaseMessage> msg, size_t port_index) override {
-    if (port_index >= num_control_ports()) {
-      throw std::runtime_error("Invalid control port index");
-    }
-
-    auto* ctrl_msg = dynamic_cast<const Message<BooleanData>*>(msg.get());
-    if (!ctrl_msg) {
-      throw std::runtime_error("Invalid control message type");
-    }
-
-    // Update last timestamp
-    control_ports_[port_index].last_timestamp = msg->time;
-    
-    if (get_control_queue(port_index).size() == max_size_per_port_) {      
-      get_control_queue(port_index).pop_front();
-    }    
+  bool equals(const Demultiplexer& other) const {
+    return Operator::equals(other);
+  }
+  
+  bool operator==(const Demultiplexer& other) const {
+    return equals(other);
+  }
 
-    // Add message to queue
-    get_control_queue(port_index).push_back(std::move(msg));
-    control_ports_with_new_data_.insert(port_index);
+  bool operator!=(const Demultiplexer& other) const {
+    return !(*this == other);
   }
 
  protected:  

libs/core/include/rtbot/FilterByValue.h

@@ -21,6 +21,10 @@ class FilterByValue : public Operator {
     add_output_port<T>();
   }
 
+  bool equals(const FilterByValue& other) const {
+    return Operator::equals(other);
+  }
+
  protected:
   void process_data() override {
     auto& input_queue = get_data_queue(0);

libs/core/include/rtbot/Input.h

@@ -20,8 +20,7 @@ class Input : public Operator {
       if (!PortType::is_valid_port_type(type)) {
         throw std::runtime_error("Unknown port type: " + type);
       }
-      PortType::add_port(*this, type, true, true);
-      last_sent_times_.push_back(0);
+      PortType::add_port(*this, type, true, false ,true);
       port_type_names_.push_back(type);
     }
   }
@@ -31,68 +30,17 @@ class Input : public Operator {
   // Get port configuration
   const std::vector<std::string>& get_port_types() const { return port_type_names_; }
 
-  // Query port state
-  bool has_sent(size_t port_index) const {
-    validate_port_index(port_index);
-    return last_sent_times_[port_index] > 0;
+  bool equals(const Input& other) const {
+    if (port_type_names_ != other.port_type_names_) return false;
+    return Operator::equals(other);
   }
-
-  timestamp_t get_last_sent_time(size_t port_index) const {
-    validate_port_index(port_index);
-    return last_sent_times_[port_index];
-  }
-
-  // State serialization
-  Bytes collect() override {
-    // First collect base state
-    Bytes bytes = Operator::collect();
-
-    // Serialize last sent times
-    size_t num_ports = last_sent_times_.size();
-    bytes.insert(bytes.end(), reinterpret_cast<const uint8_t*>(&num_ports),
-                 reinterpret_cast<const uint8_t*>(&num_ports) + sizeof(num_ports));
-
-    for (const auto& time : last_sent_times_) {
-      bytes.insert(bytes.end(), reinterpret_cast<const uint8_t*>(&time),
-                   reinterpret_cast<const uint8_t*>(&time) + sizeof(time));
-    }
-
-    // Serialize port type names
-    StateSerializer::serialize_string_vector(bytes, port_type_names_);
-
-    return bytes;
-  }
-
-  void restore(Bytes::const_iterator& it) override {
-    // First restore base state
-    Operator::restore(it);
-
-    // Restore last sent times
-    size_t num_ports = *reinterpret_cast<const size_t*>(&(*it));
-    it += sizeof(size_t);
-
-    StateSerializer::validate_port_count(num_ports, num_data_ports(), "Data");
-
-    last_sent_times_.clear();
-    last_sent_times_.reserve(num_ports);
-    for (size_t i = 0; i < num_ports; ++i) {
-      timestamp_t time = *reinterpret_cast<const timestamp_t*>(&(*it));
-      it += sizeof(timestamp_t);
-      last_sent_times_.push_back(time);
-    }
-
-    // Restore port type names
-    StateSerializer::deserialize_string_vector(it, port_type_names_);
-
-    // Validate port types match
-    if (port_type_names_.size() != num_data_ports()) {
-      throw std::runtime_error("Port type count mismatch during restore");
-    }
+  
+  bool operator==(const Input& other) const {
+    return equals(other);
   }
 
-  void reset() override {
-    Operator::reset();
-    last_sent_times_.assign(last_sent_times_.size(), 0);
+  bool operator!=(const Input& other) const {
+    return !(*this == other);
   }
 
   // Do not throw exceptions in receive_data
@@ -107,36 +55,29 @@ class Input : public Operator {
  protected:
   void process_data() override {
     // Process each port independently to allow concurrent timestamps
-    for (const auto& port_index : data_ports_with_new_data_) {
+    for (int port_index = 0; port_index < num_data_ports(); port_index++) {
       const auto& input_queue = get_data_queue(port_index);
       if (input_queue.empty()) continue;
 
       auto& output_queue = get_output_queue(port_index);
 
       // Process all messages in input queue
-      for (const auto& msg : input_queue) {
-        // Only forward if timestamp is increasing for this specific port
-        if (!has_sent(port_index) || msg->time > last_sent_times_[port_index]) {
-          output_queue.push_back(std::move(msg->clone()));
-          last_sent_times_[port_index] = msg->time;
-        }
+      for (const auto& msg : input_queue) {        
+          output_queue.push_back(std::move(msg->clone()));       
       }
 
       // Clear processed messages
       get_data_queue(port_index).clear();
     }
   }
 
-  void process_control() override {}  // No control processing needed
 
  private:
   void validate_port_index(size_t port_index) const {
     if (port_index >= num_data_ports()) {
       throw std::runtime_error("Invalid port index: " + std::to_string(port_index));
     }
   }
-
-  std::vector<timestamp_t> last_sent_times_;
   std::vector<std::string> port_type_names_;
 };