[sensor] Optimize sliding window filters to eliminate heap fragmentation (#11282)

esphome/components/sensor/__init__.py

@@ -251,6 +251,9 @@ MaxFilter = sensor_ns.class_("MaxFilter", Filter)
 SlidingWindowMovingAverageFilter = sensor_ns.class_(
     "SlidingWindowMovingAverageFilter", Filter
 )
+StreamingMinFilter = sensor_ns.class_("StreamingMinFilter", Filter)
+StreamingMaxFilter = sensor_ns.class_("StreamingMaxFilter", Filter)
+StreamingMovingAverageFilter = sensor_ns.class_("StreamingMovingAverageFilter", Filter)
 ExponentialMovingAverageFilter = sensor_ns.class_(
     "ExponentialMovingAverageFilter", Filter
 )
@@ -452,14 +455,21 @@ async def skip_initial_filter_to_code(config, filter_id):
     return cg.new_Pvariable(filter_id, config)
 
 
-@FILTER_REGISTRY.register("min", MinFilter, MIN_SCHEMA)
+@FILTER_REGISTRY.register("min", Filter, MIN_SCHEMA)
 async def min_filter_to_code(config, filter_id):
-    return cg.new_Pvariable(
-        filter_id,
-        config[CONF_WINDOW_SIZE],
-        config[CONF_SEND_EVERY],
-        config[CONF_SEND_FIRST_AT],
-    )
+    window_size: int = config[CONF_WINDOW_SIZE]
+    send_every: int = config[CONF_SEND_EVERY]
+    send_first_at: int = config[CONF_SEND_FIRST_AT]
+
+    # Optimization: Use streaming filter for batch windows (window_size == send_every)
+    # Saves 99.98% memory for large windows (e.g., 20KB → 4 bytes for window_size=5000)
+    if window_size == send_every:
+        # Use streaming filter - O(1) memory instead of O(n)
+        rhs = StreamingMinFilter.new(window_size, send_first_at)
+        return cg.Pvariable(filter_id, rhs, StreamingMinFilter)
+    # Use sliding window filter - maintains ring buffer
+    rhs = MinFilter.new(window_size, send_every, send_first_at)
+    return cg.Pvariable(filter_id, rhs, MinFilter)
 
 
 MAX_SCHEMA = cv.All(
@@ -474,14 +484,18 @@ MAX_SCHEMA = cv.All(
 )
 
 
-@FILTER_REGISTRY.register("max", MaxFilter, MAX_SCHEMA)
+@FILTER_REGISTRY.register("max", Filter, MAX_SCHEMA)
 async def max_filter_to_code(config, filter_id):
-    return cg.new_Pvariable(
-        filter_id,
-        config[CONF_WINDOW_SIZE],
-        config[CONF_SEND_EVERY],
-        config[CONF_SEND_FIRST_AT],
-    )
+    window_size: int = config[CONF_WINDOW_SIZE]
+    send_every: int = config[CONF_SEND_EVERY]
+    send_first_at: int = config[CONF_SEND_FIRST_AT]
+
+    # Optimization: Use streaming filter for batch windows (window_size == send_every)
+    if window_size == send_every:
+        rhs = StreamingMaxFilter.new(window_size, send_first_at)
+        return cg.Pvariable(filter_id, rhs, StreamingMaxFilter)
+    rhs = MaxFilter.new(window_size, send_every, send_first_at)
+    return cg.Pvariable(filter_id, rhs, MaxFilter)
 
 
 SLIDING_AVERAGE_SCHEMA = cv.All(
@@ -498,16 +512,20 @@ SLIDING_AVERAGE_SCHEMA = cv.All(
 
 @FILTER_REGISTRY.register(
     "sliding_window_moving_average",
-    SlidingWindowMovingAverageFilter,
+    Filter,
     SLIDING_AVERAGE_SCHEMA,
 )
 async def sliding_window_moving_average_filter_to_code(config, filter_id):
-    return cg.new_Pvariable(
-        filter_id,
-        config[CONF_WINDOW_SIZE],
-        config[CONF_SEND_EVERY],
-        config[CONF_SEND_FIRST_AT],
-    )
+    window_size: int = config[CONF_WINDOW_SIZE]
+    send_every: int = config[CONF_SEND_EVERY]
+    send_first_at: int = config[CONF_SEND_FIRST_AT]
+
+    # Optimization: Use streaming filter for batch windows (window_size == send_every)
+    if window_size == send_every:
+        rhs = StreamingMovingAverageFilter.new(window_size, send_first_at)
+        return cg.Pvariable(filter_id, rhs, StreamingMovingAverageFilter)
+    rhs = SlidingWindowMovingAverageFilter.new(window_size, send_every, send_first_at)
+    return cg.Pvariable(filter_id, rhs, SlidingWindowMovingAverageFilter)
 
 
 EXPONENTIAL_AVERAGE_SCHEMA = cv.All(

esphome/components/sensor/filter.cpp

@@ -32,50 +32,76 @@ void Filter::initialize(Sensor *parent, Filter *next) {
   this->next_ = next;
 }
 
-// MedianFilter
-MedianFilter::MedianFilter(size_t window_size, size_t send_every, size_t send_first_at)
-    : send_every_(send_every), send_at_(send_every - send_first_at), window_size_(window_size) {}
-void MedianFilter::set_send_every(size_t send_every) { this->send_every_ = send_every; }
-void MedianFilter::set_window_size(size_t window_size) { this->window_size_ = window_size; }
-optional<float> MedianFilter::new_value(float value) {
-  while (this->queue_.size() >= this->window_size_) {
-    this->queue_.pop_front();
-  }
-  this->queue_.push_back(value);
-  ESP_LOGVV(TAG, "MedianFilter(%p)::new_value(%f)", this, value);
+// SlidingWindowFilter
+SlidingWindowFilter::SlidingWindowFilter(size_t window_size, size_t send_every, size_t send_first_at)
+    : window_size_(window_size), send_every_(send_every), send_at_(send_every - send_first_at) {
+  // Allocate ring buffer once at initialization
+  this->window_.init(window_size);
+}
 
+optional<float> SlidingWindowFilter::new_value(float value) {
+  // Add value to ring buffer
+  if (this->window_count_ < this->window_size_) {
+    // Buffer not yet full - just append
+    this->window_.push_back(value);
+    this->window_count_++;
+  } else {
+    // Buffer full - overwrite oldest value (ring buffer)
+    this->window_[this->window_head_] = value;
+    this->window_head_++;
+    if (this->window_head_ >= this->window_size_) {
+      this->window_head_ = 0;
+    }
+  }
+
+  // Check if we should send a result
   if (++this->send_at_ >= this->send_every_) {
     this->send_at_ = 0;
-
-    float median = NAN;
-    if (!this->queue_.empty()) {
-      // Copy queue without NaN values
-      std::vector<float> median_queue;
-      median_queue.reserve(this->queue_.size());
-      for (auto v : this->queue_) {
-        if (!std::isnan(v)) {
-          median_queue.push_back(v);
-        }
-      }
-
-      sort(median_queue.begin(), median_queue.end());
-
-      size_t queue_size = median_queue.size();
-      if (queue_size) {
-        if (queue_size % 2) {
-          median = median_queue[queue_size / 2];
-        } else {
-          median = (median_queue[queue_size / 2] + median_queue[(queue_size / 2) - 1]) / 2.0f;
-        }
-      }
-    }
-
-    ESP_LOGVV(TAG, "MedianFilter(%p)::new_value(%f) SENDING %f", this, value, median);
-    return median;
+    float result = this->compute_result();
+    ESP_LOGVV(TAG, "SlidingWindowFilter(%p)::new_value(%f) SENDING %f", this, value, result);
+    return result;
   }
   return {};
 }
 
+// SortedWindowFilter
+FixedVector<float> SortedWindowFilter::get_window_values_() {
+  // Copy window without NaN values using FixedVector (no heap allocation)
+  // Returns unsorted values - caller will use std::nth_element for partial sorting as needed
+  FixedVector<float> values;
+  values.init(this->window_count_);
+  for (size_t i = 0; i < this->window_count_; i++) {
+    float v = this->window_[i];
+    if (!std::isnan(v)) {
+      values.push_back(v);
+    }
+  }
+  return values;
+}
+
+// MedianFilter
+float MedianFilter::compute_result() {
+  FixedVector<float> values = this->get_window_values_();
+  if (values.empty())
+    return NAN;
+
+  size_t size = values.size();
+  size_t mid = size / 2;
+
+  if (size % 2) {
+    // Odd number of elements - use nth_element to find middle element
+    std::nth_element(values.begin(), values.begin() + mid, values.end());
+    return values[mid];
+  }
+  // Even number of elements - need both middle elements
+  // Use nth_element to find upper middle element
+  std::nth_element(values.begin(), values.begin() + mid, values.end());
+  float upper = values[mid];
+  // Find the maximum of the lower half (which is now everything before mid)
+  float lower = *std::max_element(values.begin(), values.begin() + mid);
+  return (lower + upper) / 2.0f;
+}
+
 // SkipInitialFilter
 SkipInitialFilter::SkipInitialFilter(size_t num_to_ignore) : num_to_ignore_(num_to_ignore) {}
 optional<float> SkipInitialFilter::new_value(float value) {
@@ -91,136 +117,39 @@ optional<float> SkipInitialFilter::new_value(float value) {
 
 // QuantileFilter
 QuantileFilter::QuantileFilter(size_t window_size, size_t send_every, size_t send_first_at, float quantile)
-    : send_every_(send_every), send_at_(send_every - send_first_at), window_size_(window_size), quantile_(quantile) {}
-void QuantileFilter::set_send_every(size_t send_every) { this->send_every_ = send_every; }
-void QuantileFilter::set_window_size(size_t window_size) { this->window_size_ = window_size; }
-void QuantileFilter::set_quantile(float quantile) { this->quantile_ = quantile; }
-optional<float> QuantileFilter::new_value(float value) {
-  while (this->queue_.size() >= this->window_size_) {
-    this->queue_.pop_front();
-  }
-  this->queue_.push_back(value);
-  ESP_LOGVV(TAG, "QuantileFilter(%p)::new_value(%f), quantile:%f", this, value, this->quantile_);
+    : SortedWindowFilter(window_size, send_every, send_first_at), quantile_(quantile) {}
 
-  if (++this->send_at_ >= this->send_every_) {
-    this->send_at_ = 0;
+float QuantileFilter::compute_result() {
+  FixedVector<float> values = this->get_window_values_();
+  if (values.empty())
+    return NAN;
 
-    float result = NAN;
-    if (!this->queue_.empty()) {
-      // Copy queue without NaN values
-      std::vector<float> quantile_queue;
-      for (auto v : this->queue_) {
-        if (!std::isnan(v)) {
-          quantile_queue.push_back(v);
-        }
-      }
+  size_t position = ceilf(values.size() * this->quantile_) - 1;
+  ESP_LOGVV(TAG, "QuantileFilter(%p)::position: %zu/%zu", this, position + 1, values.size());
 
-      sort(quantile_queue.begin(), quantile_queue.end());
-
-      size_t queue_size = quantile_queue.size();
-      if (queue_size) {
-        size_t position = ceilf(queue_size * this->quantile_) - 1;
-        ESP_LOGVV(TAG, "QuantileFilter(%p)::position: %zu/%zu", this, position + 1, queue_size);
-        result = quantile_queue[position];
-      }
-    }
-
-    ESP_LOGVV(TAG, "QuantileFilter(%p)::new_value(%f) SENDING %f", this, value, result);
-    return result;
-  }
-  return {};
+  // Use nth_element to find the quantile element (O(n) instead of O(n log n))
+  std::nth_element(values.begin(), values.begin() + position, values.end());
+  return values[position];
 }
 
 // MinFilter
-MinFilter::MinFilter(size_t window_size, size_t send_every, size_t send_first_at)
-    : send_every_(send_every), send_at_(send_every - send_first_at), window_size_(window_size) {}
-void MinFilter::set_send_every(size_t send_every) { this->send_every_ = send_every; }
-void MinFilter::set_window_size(size_t window_size) { this->window_size_ = window_size; }
-optional<float> MinFilter::new_value(float value) {
-  while (this->queue_.size() >= this->window_size_) {
-    this->queue_.pop_front();
-  }
-  this->queue_.push_back(value);
-  ESP_LOGVV(TAG, "MinFilter(%p)::new_value(%f)", this, value);
-
-  if (++this->send_at_ >= this->send_every_) {
-    this->send_at_ = 0;
-
-    float min = NAN;
-    for (auto v : this->queue_) {
-      if (!std::isnan(v)) {
-        min = std::isnan(min) ? v : std::min(min, v);
-      }
-    }
-
-    ESP_LOGVV(TAG, "MinFilter(%p)::new_value(%f) SENDING %f", this, value, min);
-    return min;
-  }
-  return {};
-}
+float MinFilter::compute_result() { return this->find_extremum_<std::less<float>>(); }
 
 // MaxFilter
-MaxFilter::MaxFilter(size_t window_size, size_t send_every, size_t send_first_at)
-    : send_every_(send_every), send_at_(send_every - send_first_at), window_size_(window_size) {}
-void MaxFilter::set_send_every(size_t send_every) { this->send_every_ = send_every; }
-void MaxFilter::set_window_size(size_t window_size) { this->window_size_ = window_size; }
-optional<float> MaxFilter::new_value(float value) {
-  while (this->queue_.size() >= this->window_size_) {
-    this->queue_.pop_front();
-  }
-  this->queue_.push_back(value);
-  ESP_LOGVV(TAG, "MaxFilter(%p)::new_value(%f)", this, value);
-
-  if (++this->send_at_ >= this->send_every_) {
-    this->send_at_ = 0;
-
-    float max = NAN;
-    for (auto v : this->queue_) {
-      if (!std::isnan(v)) {
-        max = std::isnan(max) ? v : std::max(max, v);
-      }
-    }
-
-    ESP_LOGVV(TAG, "MaxFilter(%p)::new_value(%f) SENDING %f", this, value, max);
-    return max;
-  }
-  return {};
-}
+float MaxFilter::compute_result() { return this->find_extremum_<std::greater<float>>(); }
 
 // SlidingWindowMovingAverageFilter
-SlidingWindowMovingAverageFilter::SlidingWindowMovingAverageFilter(size_t window_size, size_t send_every,
-                                                                   size_t send_first_at)
-    : send_every_(send_every), send_at_(send_every - send_first_at), window_size_(window_size) {}
-void SlidingWindowMovingAverageFilter::set_send_every(size_t send_every) { this->send_every_ = send_every; }
-void SlidingWindowMovingAverageFilter::set_window_size(size_t window_size) { this->window_size_ = window_size; }
-optional<float> SlidingWindowMovingAverageFilter::new_value(float value) {
-  while (this->queue_.size() >= this->window_size_) {
-    this->queue_.pop_front();
-  }
-  this->queue_.push_back(value);
-  ESP_LOGVV(TAG, "SlidingWindowMovingAverageFilter(%p)::new_value(%f)", this, value);
-
-  if (++this->send_at_ >= this->send_every_) {
-    this->send_at_ = 0;
-
-    float sum = 0;
-    size_t valid_count = 0;
-    for (auto v : this->queue_) {
-      if (!std::isnan(v)) {
-        sum += v;
-        valid_count++;
-      }
+float SlidingWindowMovingAverageFilter::compute_result() {
+  float sum = 0;
+  size_t valid_count = 0;
+  for (size_t i = 0; i < this->window_count_; i++) {
+    float v = this->window_[i];
+    if (!std::isnan(v)) {
+      sum += v;
+      valid_count++;
     }
-
-    float average = NAN;
-    if (valid_count) {
-      average = sum / valid_count;
-    }
-
-    ESP_LOGVV(TAG, "SlidingWindowMovingAverageFilter(%p)::new_value(%f) SENDING %f", this, value, average);
-    return average;
   }
-  return {};
+  return valid_count ? sum / valid_count : NAN;
 }
 
 // ExponentialMovingAverageFilter
@@ -543,5 +472,78 @@ optional<float> ToNTCTemperatureFilter::new_value(float value) {
   return temp;
 }
 
+// StreamingFilter (base class)
+StreamingFilter::StreamingFilter(size_t window_size, size_t send_first_at)
+    : window_size_(window_size), send_first_at_(send_first_at) {}
+
+optional<float> StreamingFilter::new_value(float value) {
+  // Process the value (child class tracks min/max/sum/etc)
+  this->process_value(value);
+
+  this->count_++;
+
+  // Check if we should send (handle send_first_at for first value)
+  bool should_send = false;
+  if (this->first_send_ && this->count_ >= this->send_first_at_) {
+    should_send = true;
+    this->first_send_ = false;
+  } else if (!this->first_send_ && this->count_ >= this->window_size_) {
+    should_send = true;
+  }
+
+  if (should_send) {
+    float result = this->compute_batch_result();
+    // Reset for next batch
+    this->count_ = 0;
+    this->reset_batch();
+    ESP_LOGVV(TAG, "StreamingFilter(%p)::new_value(%f) SENDING %f", this, value, result);
+    return result;
+  }
+
+  return {};
+}
+
+// StreamingMinFilter
+void StreamingMinFilter::process_value(float value) {
+  // Update running minimum (ignore NaN values)
+  if (!std::isnan(value)) {
+    this->current_min_ = std::isnan(this->current_min_) ? value : std::min(this->current_min_, value);
+  }
+}
+
+float StreamingMinFilter::compute_batch_result() { return this->current_min_; }
+
+void StreamingMinFilter::reset_batch() { this->current_min_ = NAN; }
+
+// StreamingMaxFilter
+void StreamingMaxFilter::process_value(float value) {
+  // Update running maximum (ignore NaN values)
+  if (!std::isnan(value)) {
+    this->current_max_ = std::isnan(this->current_max_) ? value : std::max(this->current_max_, value);
+  }
+}
+
+float StreamingMaxFilter::compute_batch_result() { return this->current_max_; }
+
+void StreamingMaxFilter::reset_batch() { this->current_max_ = NAN; }
+
+// StreamingMovingAverageFilter
+void StreamingMovingAverageFilter::process_value(float value) {
+  // Accumulate sum (ignore NaN values)
+  if (!std::isnan(value)) {
+    this->sum_ += value;
+    this->valid_count_++;
+  }
+}
+
+float StreamingMovingAverageFilter::compute_batch_result() {
+  return this->valid_count_ > 0 ? this->sum_ / this->valid_count_ : NAN;
+}
+
+void StreamingMovingAverageFilter::reset_batch() {
+  this->sum_ = 0.0f;
+  this->valid_count_ = 0;
+}
+
 }  // namespace sensor
 }  // namespace esphome

esphome/components/sensor/filter.h

@@ -44,11 +44,75 @@ class Filter {
   Sensor *parent_{nullptr};
 };
 
+/** Base class for filters that use a sliding window of values.
+ *
+ * Uses a ring buffer to efficiently maintain a fixed-size sliding window without
+ * reallocations or pop_front() overhead. Eliminates deque fragmentation issues.
+ */
+class SlidingWindowFilter : public Filter {
+ public:
+  SlidingWindowFilter(size_t window_size, size_t send_every, size_t send_first_at);
+
+  optional<float> new_value(float value) final;
+
+ protected:
+  /// Called by new_value() to compute the filtered result from the current window
+  virtual float compute_result() = 0;
+
+  /// Access the sliding window values (ring buffer implementation)
+  /// Use: for (size_t i = 0; i < window_count_; i++) { float val = window_[i]; }
+  FixedVector<float> window_;
+  size_t window_head_{0};   ///< Index where next value will be written
+  size_t window_count_{0};  ///< Number of valid values in window (0 to window_size_)
+  size_t window_size_;      ///< Maximum window size
+  size_t send_every_;       ///< Send result every N values
+  size_t send_at_;          ///< Counter for send_every
+};
+
+/** Base class for Min/Max filters.
+ *
+ * Provides a templated helper to find extremum values efficiently.
+ */
+class MinMaxFilter : public SlidingWindowFilter {
+ public:
+  using SlidingWindowFilter::SlidingWindowFilter;
+
+ protected:
+  /// Helper to find min or max value in window, skipping NaN values
+  /// Usage: find_extremum_<std::less<float>>() for min, find_extremum_<std::greater<float>>() for max
+  template<typename Compare> float find_extremum_() {
+    float result = NAN;
+    Compare comp;
+    for (size_t i = 0; i < this->window_count_; i++) {
+      float v = this->window_[i];
+      if (!std::isnan(v)) {
+        result = std::isnan(result) ? v : (comp(v, result) ? v : result);
+      }
+    }
+    return result;
+  }
+};
+
+/** Base class for filters that need a sorted window (Median, Quantile).
+ *
+ * Extends SlidingWindowFilter to provide a helper that filters out NaN values.
+ * Derived classes use std::nth_element for efficient partial sorting.
+ */
+class SortedWindowFilter : public SlidingWindowFilter {
+ public:
+  using SlidingWindowFilter::SlidingWindowFilter;
+
+ protected:
+  /// Helper to get non-NaN values from the window (not sorted - caller will use nth_element)
+  /// Returns empty FixedVector if all values are NaN
+  FixedVector<float> get_window_values_();
+};
+
 /** Simple quantile filter.
  *
- * Takes the quantile of the last <send_every> values and pushes it out every <send_every>.
+ * Takes the quantile of the last <window_size> values and pushes it out every <send_every>.
  */
-class QuantileFilter : public Filter {
+class QuantileFilter : public SortedWindowFilter {
  public:
   /** Construct a QuantileFilter.
    *
@@ -61,25 +125,18 @@ class QuantileFilter : public Filter {
    */
   explicit QuantileFilter(size_t window_size, size_t send_every, size_t send_first_at, float quantile);
 
-  optional<float> new_value(float value) override;
-
-  void set_send_every(size_t send_every);
-  void set_window_size(size_t window_size);
-  void set_quantile(float quantile);
+  void set_quantile(float quantile) { this->quantile_ = quantile; }
 
  protected:
-  std::deque<float> queue_;
-  size_t send_every_;
-  size_t send_at_;
-  size_t window_size_;
+  float compute_result() override;
   float quantile_;
 };
 
 /** Simple median filter.
  *
- * Takes the median of the last <send_every> values and pushes it out every <send_every>.
+ * Takes the median of the last <window_size> values and pushes it out every <send_every>.
  */
-class MedianFilter : public Filter {
+class MedianFilter : public SortedWindowFilter {
  public:
   /** Construct a MedianFilter.
    *
@@ -89,18 +146,10 @@ class MedianFilter : public Filter {
    *   on startup being published on the first *raw* value, so with no filter applied. Must be less than or equal to
    *   send_every.
    */
-  explicit MedianFilter(size_t window_size, size_t send_every, size_t send_first_at);
-
-  optional<float> new_value(float value) override;
-
-  void set_send_every(size_t send_every);
-  void set_window_size(size_t window_size);
+  using SortedWindowFilter::SortedWindowFilter;
 
  protected:
-  std::deque<float> queue_;
-  size_t send_every_;
-  size_t send_at_;
-  size_t window_size_;
+  float compute_result() override;
 };
 
 /** Simple skip filter.
@@ -123,9 +172,9 @@ class SkipInitialFilter : public Filter {
 
 /** Simple min filter.
  *
- * Takes the min of the last <send_every> values and pushes it out every <send_every>.
+ * Takes the min of the last <window_size> values and pushes it out every <send_every>.
  */
-class MinFilter : public Filter {
+class MinFilter : public MinMaxFilter {
  public:
   /** Construct a MinFilter.
    *
@@ -135,25 +184,17 @@ class MinFilter : public Filter {
    *   on startup being published on the first *raw* value, so with no filter applied. Must be less than or equal to
    *   send_every.
    */
-  explicit MinFilter(size_t window_size, size_t send_every, size_t send_first_at);
-
-  optional<float> new_value(float value) override;
-
-  void set_send_every(size_t send_every);
-  void set_window_size(size_t window_size);
+  using MinMaxFilter::MinMaxFilter;
 
  protected:
-  std::deque<float> queue_;
-  size_t send_every_;
-  size_t send_at_;
-  size_t window_size_;
+  float compute_result() override;
 };
 
 /** Simple max filter.
  *
- * Takes the max of the last <send_every> values and pushes it out every <send_every>.
+ * Takes the max of the last <window_size> values and pushes it out every <send_every>.
  */
-class MaxFilter : public Filter {
+class MaxFilter : public MinMaxFilter {
  public:
   /** Construct a MaxFilter.
    *
@@ -163,18 +204,10 @@ class MaxFilter : public Filter {
    *   on startup being published on the first *raw* value, so with no filter applied. Must be less than or equal to
    *   send_every.
    */
-  explicit MaxFilter(size_t window_size, size_t send_every, size_t send_first_at);
-
-  optional<float> new_value(float value) override;
-
-  void set_send_every(size_t send_every);
-  void set_window_size(size_t window_size);
+  using MinMaxFilter::MinMaxFilter;
 
  protected:
-  std::deque<float> queue_;
-  size_t send_every_;
-  size_t send_at_;
-  size_t window_size_;
+  float compute_result() override;
 };
 
 /** Simple sliding window moving average filter.
@@ -182,7 +215,7 @@ class MaxFilter : public Filter {
  * Essentially just takes takes the average of the last window_size values and pushes them out
  * every send_every.
  */
-class SlidingWindowMovingAverageFilter : public Filter {
+class SlidingWindowMovingAverageFilter : public SlidingWindowFilter {
  public:
   /** Construct a SlidingWindowMovingAverageFilter.
    *
@@ -192,18 +225,10 @@ class SlidingWindowMovingAverageFilter : public Filter {
    *   on startup being published on the first *raw* value, so with no filter applied. Must be less than or equal to
    *   send_every.
    */
-  explicit SlidingWindowMovingAverageFilter(size_t window_size, size_t send_every, size_t send_first_at);
-
-  optional<float> new_value(float value) override;
-
-  void set_send_every(size_t send_every);
-  void set_window_size(size_t window_size);
+  using SlidingWindowFilter::SlidingWindowFilter;
 
  protected:
-  std::deque<float> queue_;
-  size_t send_every_;
-  size_t send_at_;
-  size_t window_size_;
+  float compute_result() override;
 };
 
 /** Simple exponential moving average filter.
@@ -476,5 +501,81 @@ class ToNTCTemperatureFilter : public Filter {
   double c_;
 };
 
+/** Base class for streaming filters (batch windows where window_size == send_every).
+ *
+ * When window_size equals send_every, we don't need a sliding window.
+ * This base class handles the common batching logic.
+ */
+class StreamingFilter : public Filter {
+ public:
+  StreamingFilter(size_t window_size, size_t send_first_at);
+
+  optional<float> new_value(float value) final;
+
+ protected:
+  /// Called by new_value() to process each value in the batch
+  virtual void process_value(float value) = 0;
+
+  /// Called by new_value() to compute the result after collecting window_size values
+  virtual float compute_batch_result() = 0;
+
+  /// Called by new_value() to reset internal state after sending a result
+  virtual void reset_batch() = 0;
+
+  size_t window_size_;
+  size_t count_{0};
+  size_t send_first_at_;
+  bool first_send_{true};
+};
+
+/** Streaming min filter for batch windows (window_size == send_every).
+ *
+ * Uses O(1) memory instead of O(n) by tracking only the minimum value.
+ */
+class StreamingMinFilter : public StreamingFilter {
+ public:
+  using StreamingFilter::StreamingFilter;
+
+ protected:
+  void process_value(float value) override;
+  float compute_batch_result() override;
+  void reset_batch() override;
+
+  float current_min_{NAN};
+};
+
+/** Streaming max filter for batch windows (window_size == send_every).
+ *
+ * Uses O(1) memory instead of O(n) by tracking only the maximum value.
+ */
+class StreamingMaxFilter : public StreamingFilter {
+ public:
+  using StreamingFilter::StreamingFilter;
+
+ protected:
+  void process_value(float value) override;
+  float compute_batch_result() override;
+  void reset_batch() override;
+
+  float current_max_{NAN};
+};
+
+/** Streaming moving average filter for batch windows (window_size == send_every).
+ *
+ * Uses O(1) memory instead of O(n) by tracking only sum and count.
+ */
+class StreamingMovingAverageFilter : public StreamingFilter {
+ public:
+  using StreamingFilter::StreamingFilter;
+
+ protected:
+  void process_value(float value) override;
+  float compute_batch_result() override;
+  void reset_batch() override;
+
+  float sum_{0.0f};
+  size_t valid_count_{0};
+};
+
 }  // namespace sensor
 }  // namespace esphome

tests/components/sensor/common.yaml

@@ -0,0 +1,101 @@
+sensor:
+  # Source sensor for testing filters
+  - platform: template
+    name: "Source Sensor"
+    id: source_sensor
+    lambda: return 42.0;
+    update_interval: 1s
+
+  # Streaming filters (window_size == send_every) - uses StreamingFilter base class
+  - platform: copy
+    source_id: source_sensor
+    name: "Streaming Min Filter"
+    filters:
+      - min:
+          window_size: 10
+          send_every: 10  # Batch window → StreamingMinFilter
+
+  - platform: copy
+    source_id: source_sensor
+    name: "Streaming Max Filter"
+    filters:
+      - max:
+          window_size: 10
+          send_every: 10  # Batch window → StreamingMaxFilter
+
+  - platform: copy
+    source_id: source_sensor
+    name: "Streaming Moving Average Filter"
+    filters:
+      - sliding_window_moving_average:
+          window_size: 10
+          send_every: 10  # Batch window → StreamingMovingAverageFilter
+
+  # Sliding window filters (window_size != send_every) - uses SlidingWindowFilter base class with ring buffer
+  - platform: copy
+    source_id: source_sensor
+    name: "Sliding Min Filter"
+    filters:
+      - min:
+          window_size: 10
+          send_every: 5  # Sliding window → MinFilter with ring buffer
+
+  - platform: copy
+    source_id: source_sensor
+    name: "Sliding Max Filter"
+    filters:
+      - max:
+          window_size: 10
+          send_every: 5  # Sliding window → MaxFilter with ring buffer
+
+  - platform: copy
+    source_id: source_sensor
+    name: "Sliding Median Filter"
+    filters:
+      - median:
+          window_size: 10
+          send_every: 5  # Sliding window → MedianFilter with ring buffer
+
+  - platform: copy
+    source_id: source_sensor
+    name: "Sliding Quantile Filter"
+    filters:
+      - quantile:
+          window_size: 10
+          send_every: 5
+          quantile: 0.9  # Sliding window → QuantileFilter with ring buffer
+
+  - platform: copy
+    source_id: source_sensor
+    name: "Sliding Moving Average Filter"
+    filters:
+      - sliding_window_moving_average:
+          window_size: 10
+          send_every: 5  # Sliding window → SlidingWindowMovingAverageFilter with ring buffer
+
+  # Edge cases
+  - platform: copy
+    source_id: source_sensor
+    name: "Large Batch Window Min"
+    filters:
+      - min:
+          window_size: 1000
+          send_every: 1000  # Large batch → StreamingMinFilter (4 bytes, not 4KB)
+
+  - platform: copy
+    source_id: source_sensor
+    name: "Small Sliding Window"
+    filters:
+      - median:
+          window_size: 3
+          send_every: 1  # Frequent output → MedianFilter with 3-element ring buffer
+
+  # send_first_at parameter test
+  - platform: copy
+    source_id: source_sensor
+    name: "Early Send Filter"
+    filters:
+      - max:
+          window_size: 10
+          send_every: 10
+          send_first_at: 1  # Send after first value

tests/components/sensor/test.esp8266-ard.yaml

@@ -0,0 +1 @@
+<<: !include common.yaml

tests/integration/README.md

@@ -7,6 +7,7 @@ This directory contains end-to-end integration tests for ESPHome, focusing on te
 - `conftest.py` - Common fixtures and utilities
 - `const.py` - Constants used throughout the integration tests
 - `types.py` - Type definitions for fixtures and functions
+- `state_utils.py` - State handling utilities (e.g., `InitialStateHelper`, `build_key_to_entity_mapping`)
 - `fixtures/` - YAML configuration files for tests
 - `test_*.py` - Individual test files
 
@@ -26,6 +27,32 @@ The `yaml_config` fixture automatically loads YAML configurations based on the t
 - `reserved_tcp_port` - Reserves a TCP port by holding the socket open until ESPHome needs it
 - `unused_tcp_port` - Provides the reserved port number for each test
 
+### Helper Utilities
+
+#### InitialStateHelper (`state_utils.py`)
+
+The `InitialStateHelper` class solves a common problem in integration tests: when an API client connects, ESPHome automatically broadcasts the current state of all entities. This can interfere with tests that want to track only new state changes triggered by test actions.
+
+**What it does:**
+- Tracks all entities (except stateless ones like buttons)
+- Swallows the first state broadcast for each entity
+- Forwards all subsequent state changes to your test callback
+- Provides `wait_for_initial_states()` to synchronize before test actions
+
+**When to use it:**
+- Any test that triggers entity state changes and needs to verify them
+- Tests that would otherwise see duplicate or unexpected states
+- Tests that need clean separation between initial state and test-triggered changes
+
+**Implementation details:**
+- Uses `(device_id, key)` tuples to uniquely identify entities across devices
+- Automatically excludes `ButtonInfo` entities (stateless)
+- Provides debug logging to track state reception (use `--log-cli-level=DEBUG`)
+- Safe for concurrent use with multiple entity types
+
+**Future work:**
+Consider converting existing integration tests to use `InitialStateHelper` for more reliable state tracking and to eliminate race conditions related to initial state broadcasts.
+
 ### Writing Tests
 
 The simplest way to write a test is to use the `run_compiled` and `api_client_connected` fixtures:
@@ -125,6 +152,54 @@ async def test_my_sensor(
 ```
 
 ##### State Subscription Pattern
+
+**Recommended: Using InitialStateHelper**
+
+When an API client connects, ESPHome automatically sends the current state of all entities. The `InitialStateHelper` (from `state_utils.py`) handles this by swallowing these initial states and only forwarding subsequent state changes to your test callback:
+
+```python
+from .state_utils import InitialStateHelper
+
+# Track state changes with futures
+loop = asyncio.get_running_loop()
+states: dict[int, EntityState] = {}
+state_future: asyncio.Future[EntityState] = loop.create_future()
+
+def on_state(state: EntityState) -> None:
+    """This callback only receives NEW state changes, not initial states."""
+    states[state.key] = state
+    # Check for specific condition using isinstance
+    if isinstance(state, SensorState) and state.state == expected_value:
+        if not state_future.done():
+            state_future.set_result(state)
+
+# Get entities and set up state synchronization
+entities, services = await client.list_entities_services()
+initial_state_helper = InitialStateHelper(entities)
+
+# Subscribe with the wrapper that filters initial states
+client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
+
+# Wait for all initial states to be broadcast
+try:
+    await initial_state_helper.wait_for_initial_states()
+except TimeoutError:
+    pytest.fail("Timeout waiting for initial states")
+
+# Now perform your test actions - on_state will only receive new changes
+# ... trigger state changes ...
+
+# Wait for expected state
+try:
+    result = await asyncio.wait_for(state_future, timeout=5.0)
+except asyncio.TimeoutError:
+    pytest.fail(f"Expected state not received. Got: {list(states.values())}")
+```
+
+**Legacy: Manual State Tracking**
+
+If you need to handle initial states manually (not recommended for new tests):
+
 ```python
 # Track state changes with futures
 loop = asyncio.get_running_loop()

tests/integration/fixtures/sensor_filters_batch_window.yaml

@@ -0,0 +1,58 @@
+esphome:
+  name: test-batch-window-filters
+
+host:
+api:
+  batch_delay: 0ms  # Disable batching to receive all state updates
+logger:
+  level: DEBUG
+
+# Template sensor that we'll use to publish values
+sensor:
+  - platform: template
+    name: "Source Sensor"
+    id: source_sensor
+    accuracy_decimals: 2
+
+  # Batch window filters (window_size == send_every) - use streaming filters
+  - platform: copy
+    source_id: source_sensor
+    name: "Min Sensor"
+    id: min_sensor
+    filters:
+      - min:
+          window_size: 5
+          send_every: 5
+          send_first_at: 1
+
+  - platform: copy
+    source_id: source_sensor
+    name: "Max Sensor"
+    id: max_sensor
+    filters:
+      - max:
+          window_size: 5
+          send_every: 5
+          send_first_at: 1
+
+  - platform: copy
+    source_id: source_sensor
+    name: "Moving Avg Sensor"
+    id: moving_avg_sensor
+    filters:
+      - sliding_window_moving_average:
+          window_size: 5
+          send_every: 5
+          send_first_at: 1
+
+# Button to trigger publishing test values
+button:
+  - platform: template
+    name: "Publish Values Button"
+    id: publish_button
+    on_press:
+      - lambda: |-
+          // Publish 10 values: 1.0, 2.0, ..., 10.0
+          for (int i = 1; i <= 10; i++) {
+            id(source_sensor).publish_state(float(i));
+          }

tests/integration/fixtures/sensor_filters_nan_handling.yaml

@@ -0,0 +1,84 @@
+esphome:
+  name: test-nan-handling
+
+host:
+api:
+  batch_delay: 0ms  # Disable batching to receive all state updates
+logger:
+  level: DEBUG
+
+sensor:
+  - platform: template
+    name: "Source NaN Sensor"
+    id: source_nan_sensor
+    accuracy_decimals: 2
+
+  - platform: copy
+    source_id: source_nan_sensor
+    name: "Min NaN Sensor"
+    id: min_nan_sensor
+    filters:
+      - min:
+          window_size: 5
+          send_every: 5
+          send_first_at: 1
+
+  - platform: copy
+    source_id: source_nan_sensor
+    name: "Max NaN Sensor"
+    id: max_nan_sensor
+    filters:
+      - max:
+          window_size: 5
+          send_every: 5
+          send_first_at: 1
+
+script:
+  - id: publish_nan_values_script
+    then:
+      - sensor.template.publish:
+          id: source_nan_sensor
+          state: 10.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_nan_sensor
+          state: !lambda 'return NAN;'
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_nan_sensor
+          state: 5.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_nan_sensor
+          state: !lambda 'return NAN;'
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_nan_sensor
+          state: 15.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_nan_sensor
+          state: 8.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_nan_sensor
+          state: !lambda 'return NAN;'
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_nan_sensor
+          state: 12.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_nan_sensor
+          state: 3.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_nan_sensor
+          state: !lambda 'return NAN;'
+
+button:
+  - platform: template
+    name: "Publish NaN Values Button"
+    id: publish_nan_button
+    on_press:
+      - script.execute: publish_nan_values_script

tests/integration/fixtures/sensor_filters_ring_buffer.yaml

@@ -0,0 +1,115 @@
+esphome:
+  name: test-sliding-window-filters
+
+host:
+api:
+  batch_delay: 0ms  # Disable batching to receive all state updates
+logger:
+  level: DEBUG
+
+# Template sensor that we'll use to publish values
+sensor:
+  - platform: template
+    name: "Source Sensor"
+    id: source_sensor
+    accuracy_decimals: 2
+
+  # ACTUAL sliding window filters (window_size != send_every) - use ring buffers
+  # Window of 5, send every 2 values
+  - platform: copy
+    source_id: source_sensor
+    name: "Sliding Min Sensor"
+    id: sliding_min_sensor
+    filters:
+      - min:
+          window_size: 5
+          send_every: 2
+          send_first_at: 1
+
+  - platform: copy
+    source_id: source_sensor
+    name: "Sliding Max Sensor"
+    id: sliding_max_sensor
+    filters:
+      - max:
+          window_size: 5
+          send_every: 2
+          send_first_at: 1
+
+  - platform: copy
+    source_id: source_sensor
+    name: "Sliding Median Sensor"
+    id: sliding_median_sensor
+    filters:
+      - median:
+          window_size: 5
+          send_every: 2
+          send_first_at: 1
+
+  - platform: copy
+    source_id: source_sensor
+    name: "Sliding Moving Avg Sensor"
+    id: sliding_moving_avg_sensor
+    filters:
+      - sliding_window_moving_average:
+          window_size: 5
+          send_every: 2
+          send_first_at: 1
+
+# Button to trigger publishing test values
+script:
+  - id: publish_values_script
+    then:
+      # Publish 10 values: 1.0, 2.0, ..., 10.0
+      # With window_size=5, send_every=2, send_first_at=1:
+      # - Output at position 1: window=[1], min=1, max=1, median=1, avg=1
+      # - Output at position 3: window=[1,2,3], min=1, max=3, median=2, avg=2
+      # - Output at position 5: window=[1,2,3,4,5], min=1, max=5, median=3, avg=3
+      # - Output at position 7: window=[3,4,5,6,7], min=3, max=7, median=5, avg=5
+      # - Output at position 9: window=[5,6,7,8,9], min=5, max=9, median=7, avg=7
+      - sensor.template.publish:
+          id: source_sensor
+          state: 1.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_sensor
+          state: 2.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_sensor
+          state: 3.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_sensor
+          state: 4.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_sensor
+          state: 5.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_sensor
+          state: 6.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_sensor
+          state: 7.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_sensor
+          state: 8.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_sensor
+          state: 9.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_sensor
+          state: 10.0
+
+button:
+  - platform: template
+    name: "Publish Values Button"
+    id: publish_button
+    on_press:
+      - script.execute: publish_values_script

tests/integration/fixtures/sensor_filters_ring_buffer_wraparound.yaml

@@ -0,0 +1,72 @@
+esphome:
+  name: test-ring-buffer-wraparound
+
+host:
+api:
+  batch_delay: 0ms  # Disable batching to receive all state updates
+logger:
+  level: DEBUG
+
+sensor:
+  - platform: template
+    name: "Source Wraparound Sensor"
+    id: source_wraparound
+    accuracy_decimals: 2
+
+  - platform: copy
+    source_id: source_wraparound
+    name: "Wraparound Min Sensor"
+    id: wraparound_min_sensor
+    filters:
+      - min:
+          window_size: 3
+          send_every: 3
+          send_first_at: 1
+
+script:
+  - id: publish_wraparound_script
+    then:
+      # Publish 9 values to test ring buffer wraparound
+      # Values: 10, 20, 30, 5, 25, 15, 40, 35, 20
+      - sensor.template.publish:
+          id: source_wraparound
+          state: 10.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_wraparound
+          state: 20.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_wraparound
+          state: 30.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_wraparound
+          state: 5.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_wraparound
+          state: 25.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_wraparound
+          state: 15.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_wraparound
+          state: 40.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_wraparound
+          state: 35.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_wraparound
+          state: 20.0
+
+button:
+  - platform: template
+    name: "Publish Wraparound Button"
+    id: publish_wraparound_button
+    on_press:
+      - script.execute: publish_wraparound_script

tests/integration/fixtures/sensor_filters_sliding_window.yaml

@@ -0,0 +1,123 @@
+esphome:
+  name: test-sliding-window-filters
+
+host:
+api:
+  batch_delay: 0ms  # Disable batching to receive all state updates
+logger:
+  level: DEBUG
+
+# Template sensor that we'll use to publish values
+sensor:
+  - platform: template
+    name: "Source Sensor"
+    id: source_sensor
+    accuracy_decimals: 2
+
+  # Min filter sensor
+  - platform: copy
+    source_id: source_sensor
+    name: "Min Sensor"
+    id: min_sensor
+    filters:
+      - min:
+          window_size: 5
+          send_every: 5
+          send_first_at: 1
+
+  # Max filter sensor
+  - platform: copy
+    source_id: source_sensor
+    name: "Max Sensor"
+    id: max_sensor
+    filters:
+      - max:
+          window_size: 5
+          send_every: 5
+          send_first_at: 1
+
+  # Median filter sensor
+  - platform: copy
+    source_id: source_sensor
+    name: "Median Sensor"
+    id: median_sensor
+    filters:
+      - median:
+          window_size: 5
+          send_every: 5
+          send_first_at: 1
+
+  # Quantile filter sensor (90th percentile)
+  - platform: copy
+    source_id: source_sensor
+    name: "Quantile Sensor"
+    id: quantile_sensor
+    filters:
+      - quantile:
+          window_size: 5
+          send_every: 5
+          send_first_at: 1
+          quantile: 0.9
+
+  # Moving average filter sensor
+  - platform: copy
+    source_id: source_sensor
+    name: "Moving Avg Sensor"
+    id: moving_avg_sensor
+    filters:
+      - sliding_window_moving_average:
+          window_size: 5
+          send_every: 5
+          send_first_at: 1
+
+# Script to publish values with delays
+script:
+  - id: publish_values_script
+    then:
+      - sensor.template.publish:
+          id: source_sensor
+          state: 1.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_sensor
+          state: 2.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_sensor
+          state: 3.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_sensor
+          state: 4.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_sensor
+          state: 5.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_sensor
+          state: 6.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_sensor
+          state: 7.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_sensor
+          state: 8.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_sensor
+          state: 9.0
+      - delay: 20ms
+      - sensor.template.publish:
+          id: source_sensor
+          state: 10.0
+
+# Button to trigger publishing test values
+button:
+  - platform: template
+    name: "Publish Values Button"
+    id: publish_button
+    on_press:
+      - script.execute: publish_values_script

tests/integration/state_utils.py

@@ -0,0 +1,167 @@
+"""Shared utilities for ESPHome integration tests - state handling."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from aioesphomeapi import ButtonInfo, EntityInfo, EntityState
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def build_key_to_entity_mapping(
+    entities: list[EntityInfo], entity_names: list[str]
+) -> dict[int, str]:
+    """Build a mapping from entity keys to entity names.
+
+    Args:
+        entities: List of entity info objects from the API
+        entity_names: List of entity names to search for in object_ids
+
+    Returns:
+        Dictionary mapping entity keys to entity names
+    """
+    key_to_entity: dict[int, str] = {}
+    for entity in entities:
+        obj_id = entity.object_id.lower()
+        for entity_name in entity_names:
+            if entity_name in obj_id:
+                key_to_entity[entity.key] = entity_name
+                break
+    return key_to_entity
+
+
+class InitialStateHelper:
+    """Helper to wait for initial states before processing test states.
+
+    When an API client connects, ESPHome sends the current state of all entities.
+    This helper wraps the user's state callback and swallows the first state for
+    each entity, then forwards all subsequent states to the user callback.
+
+    Usage:
+        entities, services = await client.list_entities_services()
+        helper = InitialStateHelper(entities)
+        client.subscribe_states(helper.on_state_wrapper(user_callback))
+        await helper.wait_for_initial_states()
+    """
+
+    def __init__(self, entities: list[EntityInfo]) -> None:
+        """Initialize the helper.
+
+        Args:
+            entities: All entities from list_entities_services()
+        """
+        # Set of (device_id, key) tuples waiting for initial state
+        # Buttons are stateless, so exclude them
+        self._wait_initial_states = {
+            (entity.device_id, entity.key)
+            for entity in entities
+            if not isinstance(entity, ButtonInfo)
+        }
+        # Keep entity info for debugging - use (device_id, key) tuple
+        self._entities_by_id = {
+            (entity.device_id, entity.key): entity for entity in entities
+        }
+
+        # Log all entities
+        _LOGGER.debug(
+            "InitialStateHelper: Found %d total entities: %s",
+            len(entities),
+            [(type(e).__name__, e.object_id) for e in entities],
+        )
+
+        # Log which ones we're waiting for
+        _LOGGER.debug(
+            "InitialStateHelper: Waiting for %d entities (excluding ButtonInfo): %s",
+            len(self._wait_initial_states),
+            [self._entities_by_id[k].object_id for k in self._wait_initial_states],
+        )
+
+        # Log which ones we're NOT waiting for
+        not_waiting = {
+            (e.device_id, e.key) for e in entities
+        } - self._wait_initial_states
+        if not_waiting:
+            not_waiting_info = [
+                f"{type(self._entities_by_id[k]).__name__}:{self._entities_by_id[k].object_id}"
+                for k in not_waiting
+            ]
+            _LOGGER.debug(
+                "InitialStateHelper: NOT waiting for %d entities: %s",
+                len(not_waiting),
+                not_waiting_info,
+            )
+
+        # Create future in the running event loop
+        self._initial_states_received = asyncio.get_running_loop().create_future()
+        # If no entities to wait for, mark complete immediately
+        if not self._wait_initial_states:
+            self._initial_states_received.set_result(True)
+
+    def on_state_wrapper(self, user_callback):
+        """Wrap a user callback to track initial states.
+
+        Args:
+            user_callback: The user's state callback function
+
+        Returns:
+            Wrapped callback that swallows first state per entity, forwards rest
+        """
+
+        def wrapper(state: EntityState) -> None:
+            """Swallow initial state per entity, forward subsequent states."""
+            # Create entity identifier tuple
+            entity_id = (state.device_id, state.key)
+
+            # Log which entity is sending state
+            if entity_id in self._entities_by_id:
+                entity = self._entities_by_id[entity_id]
+                _LOGGER.debug(
+                    "Received state for %s (type: %s, device_id: %s, key: %d)",
+                    entity.object_id,
+                    type(entity).__name__,
+                    state.device_id,
+                    state.key,
+                )
+
+            # If this entity is waiting for initial state
+            if entity_id in self._wait_initial_states:
+                # Remove from waiting set
+                self._wait_initial_states.discard(entity_id)
+
+                _LOGGER.debug(
+                    "Swallowed initial state for %s, %d entities remaining",
+                    self._entities_by_id[entity_id].object_id
+                    if entity_id in self._entities_by_id
+                    else entity_id,
+                    len(self._wait_initial_states),
+                )
+
+                # Check if we've now seen all entities
+                if (
+                    not self._wait_initial_states
+                    and not self._initial_states_received.done()
+                ):
+                    _LOGGER.debug("All initial states received")
+                    self._initial_states_received.set_result(True)
+
+                # Don't forward initial state to user
+                return
+
+            # Forward subsequent states to user callback
+            _LOGGER.debug("Forwarding state to user callback")
+            user_callback(state)
+
+        return wrapper
+
+    async def wait_for_initial_states(self, timeout: float = 5.0) -> None:
+        """Wait for all initial states to be received.
+
+        Args:
+            timeout: Maximum time to wait in seconds
+
+        Raises:
+            asyncio.TimeoutError: If initial states aren't received within timeout
+        """
+        await asyncio.wait_for(self._initial_states_received, timeout=timeout)

tests/integration/test_sensor_filters_ring_buffer.py

@@ -0,0 +1,151 @@
+"""Test sensor ring buffer filter functionality (window_size != send_every)."""
+
+from __future__ import annotations
+
+import asyncio
+
+from aioesphomeapi import EntityState, SensorState
+import pytest
+
+from .state_utils import InitialStateHelper, build_key_to_entity_mapping
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_sensor_filters_ring_buffer(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test that ring buffer filters (window_size != send_every) work correctly."""
+    loop = asyncio.get_running_loop()
+
+    # Track state changes for each sensor
+    sensor_states: dict[str, list[float]] = {
+        "sliding_min": [],
+        "sliding_max": [],
+        "sliding_median": [],
+        "sliding_moving_avg": [],
+    }
+
+    # Futures to track when we receive expected values
+    all_updates_received = loop.create_future()
+
+    def on_state(state: EntityState) -> None:
+        """Track sensor state updates."""
+        if not isinstance(state, SensorState):
+            return
+
+        # Skip NaN values
+        if state.missing_state:
+            return
+
+        # Get the sensor name from the key mapping
+        sensor_name = key_to_sensor.get(state.key)
+        if not sensor_name or sensor_name not in sensor_states:
+            return
+
+        sensor_states[sensor_name].append(state.state)
+
+        # Check if we've received enough updates from all sensors
+        # With send_every=2, send_first_at=1, we expect 5 outputs per sensor
+        if (
+            len(sensor_states["sliding_min"]) >= 5
+            and len(sensor_states["sliding_max"]) >= 5
+            and len(sensor_states["sliding_median"]) >= 5
+            and len(sensor_states["sliding_moving_avg"]) >= 5
+            and not all_updates_received.done()
+        ):
+            all_updates_received.set_result(True)
+
+    async with (
+        run_compiled(yaml_config),
+        api_client_connected() as client,
+    ):
+        # Get entities first to build key mapping
+        entities, services = await client.list_entities_services()
+
+        # Build key-to-sensor mapping
+        key_to_sensor = build_key_to_entity_mapping(
+            entities,
+            [
+                "sliding_min",
+                "sliding_max",
+                "sliding_median",
+                "sliding_moving_avg",
+            ],
+        )
+
+        # Set up initial state helper with all entities
+        initial_state_helper = InitialStateHelper(entities)
+
+        # Subscribe to state changes with wrapper
+        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
+
+        # Wait for initial states to be sent before pressing button
+        try:
+            await initial_state_helper.wait_for_initial_states()
+        except TimeoutError:
+            pytest.fail("Timeout waiting for initial states")
+
+        # Find the publish button
+        publish_button = next(
+            (e for e in entities if "publish_values_button" in e.object_id.lower()),
+            None,
+        )
+        assert publish_button is not None, "Publish Values Button not found"
+
+        # Press the button to publish test values
+        client.button_command(publish_button.key)
+
+        # Wait for all sensors to receive their values
+        try:
+            await asyncio.wait_for(all_updates_received, timeout=10.0)
+        except TimeoutError:
+            # Provide detailed failure info
+            pytest.fail(
+                f"Timeout waiting for updates. Received states:\n"
+                f"  min: {sensor_states['sliding_min']}\n"
+                f"  max: {sensor_states['sliding_max']}\n"
+                f"  median: {sensor_states['sliding_median']}\n"
+                f"  moving_avg: {sensor_states['sliding_moving_avg']}"
+            )
+
+        # Verify we got 5 outputs per sensor (positions 1, 3, 5, 7, 9)
+        assert len(sensor_states["sliding_min"]) == 5, (
+            f"Min sensor should have 5 values, got {len(sensor_states['sliding_min'])}: {sensor_states['sliding_min']}"
+        )
+        assert len(sensor_states["sliding_max"]) == 5
+        assert len(sensor_states["sliding_median"]) == 5
+        assert len(sensor_states["sliding_moving_avg"]) == 5
+
+        # Verify the values at each output position
+        # Position 1: window=[1]
+        assert sensor_states["sliding_min"][0] == pytest.approx(1.0)
+        assert sensor_states["sliding_max"][0] == pytest.approx(1.0)
+        assert sensor_states["sliding_median"][0] == pytest.approx(1.0)
+        assert sensor_states["sliding_moving_avg"][0] == pytest.approx(1.0)
+
+        # Position 3: window=[1,2,3]
+        assert sensor_states["sliding_min"][1] == pytest.approx(1.0)
+        assert sensor_states["sliding_max"][1] == pytest.approx(3.0)
+        assert sensor_states["sliding_median"][1] == pytest.approx(2.0)
+        assert sensor_states["sliding_moving_avg"][1] == pytest.approx(2.0)
+
+        # Position 5: window=[1,2,3,4,5]
+        assert sensor_states["sliding_min"][2] == pytest.approx(1.0)
+        assert sensor_states["sliding_max"][2] == pytest.approx(5.0)
+        assert sensor_states["sliding_median"][2] == pytest.approx(3.0)
+        assert sensor_states["sliding_moving_avg"][2] == pytest.approx(3.0)
+
+        # Position 7: window=[3,4,5,6,7] (ring buffer wrapped)
+        assert sensor_states["sliding_min"][3] == pytest.approx(3.0)
+        assert sensor_states["sliding_max"][3] == pytest.approx(7.0)
+        assert sensor_states["sliding_median"][3] == pytest.approx(5.0)
+        assert sensor_states["sliding_moving_avg"][3] == pytest.approx(5.0)
+
+        # Position 9: window=[5,6,7,8,9] (ring buffer wrapped)
+        assert sensor_states["sliding_min"][4] == pytest.approx(5.0)
+        assert sensor_states["sliding_max"][4] == pytest.approx(9.0)
+        assert sensor_states["sliding_median"][4] == pytest.approx(7.0)
+        assert sensor_states["sliding_moving_avg"][4] == pytest.approx(7.0)

tests/integration/test_sensor_filters_sliding_window.py

@@ -0,0 +1,395 @@
+"""Test sensor sliding window filter functionality."""
+
+from __future__ import annotations
+
+import asyncio
+
+from aioesphomeapi import EntityState, SensorState
+import pytest
+
+from .state_utils import InitialStateHelper, build_key_to_entity_mapping
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_sensor_filters_sliding_window(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test that sliding window filters (min, max, median, quantile, moving_average) work correctly."""
+    loop = asyncio.get_running_loop()
+
+    # Track state changes for each sensor
+    sensor_states: dict[str, list[float]] = {
+        "min_sensor": [],
+        "max_sensor": [],
+        "median_sensor": [],
+        "quantile_sensor": [],
+        "moving_avg_sensor": [],
+    }
+
+    # Futures to track when we receive expected values
+    min_received = loop.create_future()
+    max_received = loop.create_future()
+    median_received = loop.create_future()
+    quantile_received = loop.create_future()
+    moving_avg_received = loop.create_future()
+
+    def on_state(state: EntityState) -> None:
+        """Track sensor state updates."""
+        if not isinstance(state, SensorState):
+            return
+
+        # Skip NaN values
+        if state.missing_state:
+            return
+
+        # Get the sensor name from the key mapping
+        sensor_name = key_to_sensor.get(state.key)
+        if not sensor_name or sensor_name not in sensor_states:
+            return
+
+        sensor_states[sensor_name].append(state.state)
+
+        # Check if we received the expected final value
+        # After publishing 10 values [1.0, 2.0, ..., 10.0], the window has the last 5: [2, 3, 4, 5, 6]
+        # Filters send at position 1 and position 6 (send_every=5 means every 5th value after first)
+        if (
+            sensor_name == "min_sensor"
+            and state.state == pytest.approx(2.0)
+            and not min_received.done()
+        ):
+            min_received.set_result(True)
+        elif (
+            sensor_name == "max_sensor"
+            and state.state == pytest.approx(6.0)
+            and not max_received.done()
+        ):
+            max_received.set_result(True)
+        elif (
+            sensor_name == "median_sensor"
+            and state.state == pytest.approx(4.0)
+            and not median_received.done()
+        ):
+            # Median of [2, 3, 4, 5, 6] = 4
+            median_received.set_result(True)
+        elif (
+            sensor_name == "quantile_sensor"
+            and state.state == pytest.approx(6.0)
+            and not quantile_received.done()
+        ):
+            # 90th percentile of [2, 3, 4, 5, 6] = 6
+            quantile_received.set_result(True)
+        elif (
+            sensor_name == "moving_avg_sensor"
+            and state.state == pytest.approx(4.0)
+            and not moving_avg_received.done()
+        ):
+            # Average of [2, 3, 4, 5, 6] = 4
+            moving_avg_received.set_result(True)
+
+    async with (
+        run_compiled(yaml_config),
+        api_client_connected() as client,
+    ):
+        # Get entities first to build key mapping
+        entities, services = await client.list_entities_services()
+
+        # Build key-to-sensor mapping
+        key_to_sensor = build_key_to_entity_mapping(
+            entities,
+            [
+                "min_sensor",
+                "max_sensor",
+                "median_sensor",
+                "quantile_sensor",
+                "moving_avg_sensor",
+            ],
+        )
+
+        # Set up initial state helper with all entities
+        initial_state_helper = InitialStateHelper(entities)
+
+        # Subscribe to state changes with wrapper
+        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
+
+        # Wait for initial states to be sent before pressing button
+        try:
+            await initial_state_helper.wait_for_initial_states()
+        except TimeoutError:
+            pytest.fail("Timeout waiting for initial states")
+
+        # Find the publish button
+        publish_button = next(
+            (e for e in entities if "publish_values_button" in e.object_id.lower()),
+            None,
+        )
+        assert publish_button is not None, "Publish Values Button not found"
+
+        # Press the button to publish test values
+        client.button_command(publish_button.key)
+
+        # Wait for all sensors to receive their final values
+        try:
+            await asyncio.wait_for(
+                asyncio.gather(
+                    min_received,
+                    max_received,
+                    median_received,
+                    quantile_received,
+                    moving_avg_received,
+                ),
+                timeout=10.0,
+            )
+        except TimeoutError:
+            # Provide detailed failure info
+            pytest.fail(
+                f"Timeout waiting for expected values. Received states:\n"
+                f"  min: {sensor_states['min_sensor']}\n"
+                f"  max: {sensor_states['max_sensor']}\n"
+                f"  median: {sensor_states['median_sensor']}\n"
+                f"  quantile: {sensor_states['quantile_sensor']}\n"
+                f"  moving_avg: {sensor_states['moving_avg_sensor']}"
+            )
+
+        # Verify we got the expected values
+        # With batch_delay: 0ms, we should receive all outputs
+        # Filters output at positions 1 and 6 (send_every: 5)
+        assert len(sensor_states["min_sensor"]) == 2, (
+            f"Min sensor should have 2 values, got {len(sensor_states['min_sensor'])}: {sensor_states['min_sensor']}"
+        )
+        assert len(sensor_states["max_sensor"]) == 2, (
+            f"Max sensor should have 2 values, got {len(sensor_states['max_sensor'])}: {sensor_states['max_sensor']}"
+        )
+        assert len(sensor_states["median_sensor"]) == 2
+        assert len(sensor_states["quantile_sensor"]) == 2
+        assert len(sensor_states["moving_avg_sensor"]) == 2
+
+        # Verify the first output (after 1 value: [1])
+        assert sensor_states["min_sensor"][0] == pytest.approx(1.0), (
+            f"First min should be 1.0, got {sensor_states['min_sensor'][0]}"
+        )
+        assert sensor_states["max_sensor"][0] == pytest.approx(1.0), (
+            f"First max should be 1.0, got {sensor_states['max_sensor'][0]}"
+        )
+        assert sensor_states["median_sensor"][0] == pytest.approx(1.0), (
+            f"First median should be 1.0, got {sensor_states['median_sensor'][0]}"
+        )
+        assert sensor_states["moving_avg_sensor"][0] == pytest.approx(1.0), (
+            f"First moving avg should be 1.0, got {sensor_states['moving_avg_sensor'][0]}"
+        )
+
+        # Verify the second output (after 6 values, window has [2, 3, 4, 5, 6])
+        assert sensor_states["min_sensor"][1] == pytest.approx(2.0), (
+            f"Second min should be 2.0, got {sensor_states['min_sensor'][1]}"
+        )
+        assert sensor_states["max_sensor"][1] == pytest.approx(6.0), (
+            f"Second max should be 6.0, got {sensor_states['max_sensor'][1]}"
+        )
+        assert sensor_states["median_sensor"][1] == pytest.approx(4.0), (
+            f"Second median should be 4.0, got {sensor_states['median_sensor'][1]}"
+        )
+        assert sensor_states["moving_avg_sensor"][1] == pytest.approx(4.0), (
+            f"Second moving avg should be 4.0, got {sensor_states['moving_avg_sensor'][1]}"
+        )
+
+
+@pytest.mark.asyncio
+async def test_sensor_filters_nan_handling(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test that sliding window filters handle NaN values correctly."""
+    loop = asyncio.get_running_loop()
+
+    # Track states
+    min_states: list[float] = []
+    max_states: list[float] = []
+
+    # Future to track completion
+    filters_completed = loop.create_future()
+
+    def on_state(state: EntityState) -> None:
+        """Track sensor state updates."""
+        if not isinstance(state, SensorState):
+            return
+
+        # Skip NaN values
+        if state.missing_state:
+            return
+
+        sensor_name = key_to_sensor.get(state.key)
+
+        if sensor_name == "min_nan":
+            min_states.append(state.state)
+        elif sensor_name == "max_nan":
+            max_states.append(state.state)
+
+        # Check if both have received their final values
+        # With batch_delay: 0ms, we should receive 2 outputs each
+        if (
+            len(min_states) >= 2
+            and len(max_states) >= 2
+            and not filters_completed.done()
+        ):
+            filters_completed.set_result(True)
+
+    async with (
+        run_compiled(yaml_config),
+        api_client_connected() as client,
+    ):
+        # Get entities first to build key mapping
+        entities, services = await client.list_entities_services()
+
+        # Build key-to-sensor mapping
+        key_to_sensor = build_key_to_entity_mapping(entities, ["min_nan", "max_nan"])
+
+        # Set up initial state helper with all entities
+        initial_state_helper = InitialStateHelper(entities)
+
+        # Subscribe to state changes with wrapper
+        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
+
+        # Wait for initial states
+        try:
+            await initial_state_helper.wait_for_initial_states()
+        except TimeoutError:
+            pytest.fail("Timeout waiting for initial states")
+
+        # Find the publish button
+        publish_button = next(
+            (e for e in entities if "publish_nan_values_button" in e.object_id.lower()),
+            None,
+        )
+        assert publish_button is not None, "Publish NaN Values Button not found"
+
+        # Press the button
+        client.button_command(publish_button.key)
+
+        # Wait for filters to process
+        try:
+            await asyncio.wait_for(filters_completed, timeout=10.0)
+        except TimeoutError:
+            pytest.fail(
+                f"Timeout waiting for NaN handling. Received:\n"
+                f"  min_states: {min_states}\n"
+                f"  max_states: {max_states}"
+            )
+
+        # Verify NaN values were ignored
+        # With batch_delay: 0ms, we should receive both outputs (at positions 1 and 6)
+        # Position 1: window=[10], min=10, max=10
+        # Position 6: window=[NaN, 5, NaN, 15, 8], ignoring NaN -> [5, 15, 8], min=5, max=15
+        assert len(min_states) == 2, (
+            f"Should have 2 min states, got {len(min_states)}: {min_states}"
+        )
+        assert len(max_states) == 2, (
+            f"Should have 2 max states, got {len(max_states)}: {max_states}"
+        )
+
+        # First output
+        assert min_states[0] == pytest.approx(10.0), (
+            f"First min should be 10.0, got {min_states[0]}"
+        )
+        assert max_states[0] == pytest.approx(10.0), (
+            f"First max should be 10.0, got {max_states[0]}"
+        )
+
+        # Second output - verify NaN values were ignored
+        assert min_states[1] == pytest.approx(5.0), (
+            f"Second min should ignore NaN and return 5.0, got {min_states[1]}"
+        )
+        assert max_states[1] == pytest.approx(15.0), (
+            f"Second max should ignore NaN and return 15.0, got {max_states[1]}"
+        )
+
+
+@pytest.mark.asyncio
+async def test_sensor_filters_ring_buffer_wraparound(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test that ring buffer correctly wraps around when window fills up."""
+    loop = asyncio.get_running_loop()
+
+    min_states: list[float] = []
+
+    test_completed = loop.create_future()
+
+    def on_state(state: EntityState) -> None:
+        """Track min sensor states."""
+        if not isinstance(state, SensorState):
+            return
+
+        # Skip NaN values
+        if state.missing_state:
+            return
+
+        sensor_name = key_to_sensor.get(state.key)
+
+        if sensor_name == "wraparound_min":
+            min_states.append(state.state)
+            # With batch_delay: 0ms, we should receive all 3 outputs
+            if len(min_states) >= 3 and not test_completed.done():
+                test_completed.set_result(True)
+
+    async with (
+        run_compiled(yaml_config),
+        api_client_connected() as client,
+    ):
+        # Get entities first to build key mapping
+        entities, services = await client.list_entities_services()
+
+        # Build key-to-sensor mapping
+        key_to_sensor = build_key_to_entity_mapping(entities, ["wraparound_min"])
+
+        # Set up initial state helper with all entities
+        initial_state_helper = InitialStateHelper(entities)
+
+        # Subscribe to state changes with wrapper
+        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
+
+        # Wait for initial state
+        try:
+            await initial_state_helper.wait_for_initial_states()
+        except TimeoutError:
+            pytest.fail("Timeout waiting for initial state")
+
+        # Find the publish button
+        publish_button = next(
+            (e for e in entities if "publish_wraparound_button" in e.object_id.lower()),
+            None,
+        )
+        assert publish_button is not None, "Publish Wraparound Button not found"
+
+        # Press the button
+        # Will publish: 10, 20, 30, 5, 25, 15, 40, 35, 20
+        client.button_command(publish_button.key)
+
+        # Wait for completion
+        try:
+            await asyncio.wait_for(test_completed, timeout=10.0)
+        except TimeoutError:
+            pytest.fail(f"Timeout waiting for wraparound test. Received: {min_states}")
+
+        # Verify outputs
+        # With window_size=3, send_every=3, we get outputs at positions 1, 4, 7
+        # Position 1: window=[10], min=10
+        # Position 4: window=[20, 30, 5], min=5
+        # Position 7: window=[15, 40, 35], min=15
+        # With batch_delay: 0ms, we should receive all 3 outputs
+        assert len(min_states) == 3, (
+            f"Should have 3 states, got {len(min_states)}: {min_states}"
+        )
+        assert min_states[0] == pytest.approx(10.0), (
+            f"First min should be 10.0, got {min_states[0]}"
+        )
+        assert min_states[1] == pytest.approx(5.0), (
+            f"Second min should be 5.0, got {min_states[1]}"
+        )
+        assert min_states[2] == pytest.approx(15.0), (
+            f"Third min should be 15.0, got {min_states[2]}"
+        )