Merge pull request #3620 from bramstroker/feat/outlier-filter

Implement outlier filter for energy sensors

Author: bramstroker

custom_components/powercalc/config_flow.py

@@ -67,6 +67,7 @@
 )
 from .flow_helper.schema import (
     SCHEMA_ENERGY_SENSOR_TOGGLE,
+    SCHEMA_SENSOR_ENERGY_OPTIONS,
     SCHEMA_UTILITY_METER_OPTIONS,
     SCHEMA_UTILITY_METER_TOGGLE,
 )
@@ -300,6 +301,17 @@ async def async_step_utility_meter_options(self, user_input: dict[str, Any] | No
             user_input,
         )
 
+    async def async_step_energy_options(self, user_input: dict[str, Any] | None = None) -> FlowResult:
+        """Handle the flow for utility meter options."""
+        return await self.handle_form_step(
+            PowercalcFormStep(
+                step=Step.ENERGY_OPTIONS,
+                schema=SCHEMA_SENSOR_ENERGY_OPTIONS,
+                continue_utility_meter_options_step=not self.is_options_flow,
+            ),
+            user_input,
+        )
+
 
 class PowercalcConfigFlow(PowercalcCommonFlow, ConfigFlow, domain=DOMAIN):
     """Handle a config flow for PowerCalc."""
@@ -486,7 +498,7 @@ def build_menu(self) -> list[Step]:
         if self.selected_sensor_type == SensorType.DAILY_ENERGY:
             menu.append(Step.DAILY_ENERGY)
         if self.selected_sensor_type == SensorType.REAL_POWER:
-            menu.append(Step.REAL_POWER)
+            menu.extend([Step.REAL_POWER, Step.ENERGY_OPTIONS])
         if self.selected_sensor_type == SensorType.GROUP:
             menu.extend(self.flow_handlers[FlowType.GROUP].build_group_menu())
 

custom_components/powercalc/const.py

@@ -72,6 +72,8 @@
 CONF_ENERGY_SENSOR_PRECISION = "energy_sensor_precision"
 CONF_ENERGY_SENSOR_UNIT_PREFIX = "energy_sensor_unit_prefix"
 CONF_ENERGY_UPDATE_INTERVAL = "energy_update_interval"
+CONF_ENERGY_FILTER_OUTLIER_ENABLED = "energy_filter_outlier_enabled"
+CONF_ENERGY_FILTER_OUTLIER_MAX = "energy_filter_outlier_max_step"
 CONF_EXCLUDE_ENTITIES = "exclude_entities"
 CONF_FILTER = "filter"
 CONF_FIXED = "fixed"

custom_components/powercalc/filter/__init__.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+from collections import deque
+import statistics
+
+
+class OutlierFilter:
+    """Simple rolling-window outlier filter using median + MAD.
+
+    - Warm-up: accepts the first `min_samples` values unconditionally.
+    - After that: rejects values whose modified Z-score > `max_z_score`.
+    """
+
+    def __init__(
+        self,
+        window_size: int = 30,
+        min_samples: int = 10,
+        max_z_score: float = 5.0,
+        max_expected_step: int = 1000,
+    ) -> None:
+        self._window_size = window_size
+        self._min_samples = min_samples
+        self._max_z_score = max_z_score
+        self._values: deque[float] = deque(maxlen=window_size)
+        self._max_expected_step = max_expected_step
+
+    @property
+    def values(self) -> list[float]:
+        return list(self._values)
+
+    def _is_outlier(self, value: float) -> bool:
+        """Return True if value is considered an outlier."""
+        if len(self._values) < self._min_samples:
+            return False
+
+        median = statistics.median(self._values)
+
+        # 1) Always allow downward transitions (light turning OFF)
+        if value <= median:
+            return False
+
+        # 2) Allow reasonable upward transitions (light turning ON)
+        # e.g. below 200 W difference - always accept
+        if value - median < self._max_expected_step:
+            return False
+
+        # 3) For larger jumps, use proper outlier detection (MAD)
+        abs_devs = [abs(x - median) for x in self._values]
+        mad = statistics.median(abs_devs) or 0
+
+        if mad == 0:
+            return False  # pragma: no cover
+
+        z = 0.6745 * (value - median) / mad
+        return abs(z) > self._max_z_score
+
+    def accept(self, value: float) -> bool:
+        """Return True if value should be accepted (not an outlier).
+
+        Also updates the internal window if the value is accepted.
+        """
+        if self._is_outlier(value):
+            return False
+
+        self._values.append(value)
+        return True

custom_components/powercalc/filter/outlier.py

@@ -40,6 +40,7 @@ class Step(StrEnum):
     USER = "user"
     SMART_SWITCH = "smart_switch"
     INIT = "init"
+    ENERGY_OPTIONS = "energy_options"
     UTILITY_METER_OPTIONS = "utility_meter_options"
     GLOBAL_CONFIGURATION = "global_configuration"
     GLOBAL_CONFIGURATION_DISCOVERY = "global_configuration_discovery"

custom_components/powercalc/flow_helper/common.py

@@ -47,7 +47,7 @@ async def async_step_real_power(self, user_input: dict[str, Any] | None = None)
             PowercalcFormStep(
                 step=Step.REAL_POWER,
                 schema=SCHEMA_REAL_POWER,
-                continue_utility_meter_options_step=True,
+                next_step=Step.ENERGY_OPTIONS,
             ),
             user_input,
         )

custom_components/powercalc/flow_helper/flows/real_power.py

@@ -40,7 +40,11 @@
 from custom_components.powercalc.flow_helper.common import FlowType, PowercalcFormStep, Step, fill_schema_defaults
 from custom_components.powercalc.flow_helper.flows.global_configuration import get_global_powercalc_config
 from custom_components.powercalc.flow_helper.flows.library import SCHEMA_POWER_OPTIONS_LIBRARY, SCHEMA_POWER_SMART_SWITCH
-from custom_components.powercalc.flow_helper.schema import SCHEMA_ENERGY_OPTIONS, SCHEMA_ENERGY_SENSOR_TOGGLE, SCHEMA_UTILITY_METER_TOGGLE
+from custom_components.powercalc.flow_helper.schema import (
+    SCHEMA_ENERGY_SENSOR_TOGGLE,
+    SCHEMA_SENSOR_ENERGY_OPTIONS,
+    SCHEMA_UTILITY_METER_TOGGLE,
+)
 from custom_components.powercalc.power_profile.power_profile import DeviceType
 from custom_components.powercalc.strategy.wled import CONFIG_SCHEMA as SCHEMA_POWER_WLED
 
@@ -243,7 +247,7 @@ async def async_step_power_advanced(self, user_input: dict[str, Any] | None = No
 
         schema = SCHEMA_POWER_ADVANCED
         if self.flow.sensor_config.get(CONF_CREATE_ENERGY_SENSOR):
-            schema = schema.extend(SCHEMA_ENERGY_OPTIONS.schema)
+            schema = schema.extend(SCHEMA_SENSOR_ENERGY_OPTIONS.schema)
 
         return await self.flow.handle_form_step(
             PowercalcFormStep(

custom_components/powercalc/flow_helper/flows/virtual_power.py

@@ -5,6 +5,8 @@
 from custom_components.powercalc.const import (
     CONF_CREATE_ENERGY_SENSOR,
     CONF_CREATE_UTILITY_METERS,
+    CONF_ENERGY_FILTER_OUTLIER_ENABLED,
+    CONF_ENERGY_FILTER_OUTLIER_MAX,
     CONF_ENERGY_INTEGRATION_METHOD,
     CONF_ENERGY_SENSOR_UNIT_PREFIX,
     CONF_SUB_PROFILE,
@@ -55,6 +57,15 @@
     },
 )
 
+SCHEMA_SENSOR_ENERGY_OPTIONS = SCHEMA_ENERGY_OPTIONS.extend(
+    vol.Schema(
+        {
+            vol.Optional(CONF_ENERGY_FILTER_OUTLIER_ENABLED, default=False): selector.BooleanSelector(),
+            vol.Optional(CONF_ENERGY_FILTER_OUTLIER_MAX): selector.NumberSelector(),
+        },
+    ).schema,
+)
+
 
 SCHEMA_UTILITY_METER_OPTIONS = vol.Schema(
     {

custom_components/powercalc/flow_helper/schema.py

@@ -57,6 +57,8 @@
     CONF_DAILY_FIXED_ENERGY,
     CONF_DELAY,
     CONF_DISABLE_STANDBY_POWER,
+    CONF_ENERGY_FILTER_OUTLIER_ENABLED,
+    CONF_ENERGY_FILTER_OUTLIER_MAX,
     CONF_ENERGY_INTEGRATION_METHOD,
     CONF_ENERGY_SENSOR_CATEGORY,
     CONF_ENERGY_SENSOR_ID,
@@ -195,6 +197,8 @@
     vol.Optional(CONF_ENERGY_SENSOR_NAMING): validate_name_pattern,
     vol.Optional(CONF_ENERGY_SENSOR_CATEGORY): vol.In(ENTITY_CATEGORIES),
     vol.Optional(CONF_ENERGY_INTEGRATION_METHOD): vol.In(ENERGY_INTEGRATION_METHODS),
+    vol.Optional(CONF_ENERGY_FILTER_OUTLIER_ENABLED): cv.boolean,
+    vol.Optional(CONF_ENERGY_FILTER_OUTLIER_MAX): cv.positive_int,
     vol.Optional(CONF_ENERGY_SENSOR_UNIT_PREFIX): vol.In([cls.value for cls in UnitPrefix]),
     vol.Optional(CONF_CREATE_GROUP): cv.string,
     vol.Optional(CONF_GROUP_ENERGY_START_AT_ZERO): cv.boolean,

custom_components/powercalc/sensor.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from datetime import timedelta
+from datetime import datetime, timedelta
 from decimal import Decimal
 import inspect
 import logging
@@ -10,11 +10,13 @@
 from homeassistant.const import (
     ATTR_UNIT_OF_MEASUREMENT,
     CONF_NAME,
+    STATE_UNAVAILABLE,
+    STATE_UNKNOWN,
     UnitOfEnergy,
     UnitOfPower,
     UnitOfTime,
 )
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HomeAssistant, State, callback
 from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.entity import EntityCategory
 import homeassistant.helpers.entity_registry as er
@@ -25,6 +27,8 @@
     ATTR_SOURCE_DOMAIN,
     ATTR_SOURCE_ENTITY,
     CONF_DISABLE_EXTENDED_ATTRIBUTES,
+    CONF_ENERGY_FILTER_OUTLIER_ENABLED,
+    CONF_ENERGY_FILTER_OUTLIER_MAX,
     CONF_ENERGY_INTEGRATION_METHOD,
     CONF_ENERGY_SENSOR_CATEGORY,
     CONF_ENERGY_SENSOR_ID,
@@ -40,6 +44,7 @@
 )
 from custom_components.powercalc.device_binding import get_device_info
 from custom_components.powercalc.errors import SensorConfigurationError
+from custom_components.powercalc.filter.outlier import OutlierFilter
 
 from .abstract import (
     BaseEntity,
@@ -280,6 +285,34 @@ def __init__(
         self._attr_suggested_display_precision = round_digits
         if entity_category:
             self._attr_entity_category = EntityCategory(entity_category)
+        self._filter_outliers = bool(sensor_config.get(CONF_ENERGY_FILTER_OUTLIER_ENABLED, False))
+        self._outlier_filter = OutlierFilter(
+            window_size=30,
+            min_samples=5,
+            max_z_score=3.5,
+            max_expected_step=sensor_config.get(CONF_ENERGY_FILTER_OUTLIER_MAX, 1000),
+        )
+
+    def _integrate_on_state_change(
+        self,
+        old_timestamp: datetime | None,
+        new_timestamp: datetime | None,
+        old_state: State | None,
+        new_state: State | None,
+    ) -> None:
+        """Override to add outlier filtering."""
+
+        if self._filter_outliers and new_state is not None:
+            valid_state = new_state.state not in (STATE_UNKNOWN, STATE_UNAVAILABLE)
+            if valid_state and not self._outlier_filter.accept(float(new_state.state)):
+                _LOGGER.debug(
+                    "%s: Rejecting power value %s as outlier for energy integration",
+                    self.entity_id,
+                    new_state.state,
+                )
+                return
+
+        super()._integrate_on_state_change(old_timestamp, new_timestamp, old_state, new_state)
 
     @property
     def extra_state_attributes(self) -> dict[str, str] | None: