Enable setting a median manually for the band-pass (outlier) filter

homeassistant/components/sensor/filter.py

@@ -8,6 +8,7 @@
 import statistics
 from collections import deque, Counter
 from numbers import Number
+import math
 
 import voluptuous as vol
 
@@ -35,6 +36,7 @@
 CONF_FILTER_WINDOW_SIZE = 'window_size'
 CONF_FILTER_PRECISION = 'precision'
 CONF_FILTER_RADIUS = 'radius'
+CONF_FILTER_MEDIAN = 'median'
 CONF_FILTER_TIME_CONSTANT = 'time_constant'
 CONF_TIME_SMA_TYPE = 'type'
 
@@ -44,6 +46,7 @@
 DEFAULT_PRECISION = 2
 DEFAULT_FILTER_RADIUS = 2.0
 DEFAULT_FILTER_TIME_CONSTANT = 10
+NO_MANUAL_MEDIAN = math.inf
 
 NAME_TEMPLATE = "{} filter"
 ICON = 'mdi:chart-line-variant'
@@ -60,6 +63,8 @@
                  default=DEFAULT_WINDOW_SIZE): vol.Coerce(int),
     vol.Optional(CONF_FILTER_RADIUS,
                  default=DEFAULT_FILTER_RADIUS): vol.Coerce(float),
+    vol.Optional(CONF_FILTER_MEDIAN,
+                 default=NO_MANUAL_MEDIAN): vol.Coerce(float),
 })
 
 FILTER_LOWPASS_SCHEMA = FILTER_SCHEMA.extend({
@@ -244,28 +249,35 @@ class OutlierFilter(Filter):
 
     Args:
         radius (float): band radius
+        median (float): band center
+            (defaults to NO_MANUAL_MEDIAN for automatic computation)
     """
 
-    def __init__(self, window_size, precision, entity, radius):
+    def __init__(self, window_size, precision, entity, radius,
+                 median=NO_MANUAL_MEDIAN):
         """Initialize Filter."""
         super().__init__(FILTER_NAME_OUTLIER, window_size, precision, entity)
         self._radius = radius
+        self._manual_median = median
         self._stats_internal = Counter()
 
     def _filter_state(self, new_state):
         """Implement the outlier filter."""
         new_state = float(new_state)
 
-        if (self.states and
-                abs(new_state - statistics.median(self.states))
-                > self._radius):
+        if self.states:
+            if self._manual_median != NO_MANUAL_MEDIAN:
+                diff = abs(new_state - self._manual_median)
+            else:
+                diff = abs(new_state - statistics.median(self.states))
+            if diff > self._radius:
 
-            self._stats_internal['erasures'] += 1
+                self._stats_internal['erasures'] += 1
 
-            _LOGGER.debug("Outlier nr. %s in %s: %s",
-                          self._stats_internal['erasures'],
-                          self._entity, new_state)
-            return self.states[-1]
+                _LOGGER.debug("Outlier nr. %s in %s: %s",
+                              self._stats_internal['erasures'],
+                              self._entity, new_state)
+                return self.states[-1]
         return new_state
 
 

tests/components/sensor/test_filter.py

@@ -72,6 +72,17 @@ def test_outlier(self):
             filtered = filt.filter_state(state)
         self.assertEqual(22, filtered)
 
+    def test_outlier_median(self):
+        """Test if outlier filter with manual median works."""
+        filt = OutlierFilter(window_size=10,
+                             precision=2,
+                             entity=None,
+                             radius=4.0,
+                             median=0)
+        for state in self.values:
+            filtered = filt.filter_state(state)
+        self.assertEqual(0, filtered)
+
     def test_lowpass(self):
         """Test if lowpass filter works."""
         filt = LowPassFilter(window_size=10,