Merge pull request #2876 from bramstroker/feat/multi-switch-auto-enti…

…ties

Implement auto populate multi switch entities

README.md

@@ -11,7 +11,7 @@
 
 # <img src="https://docs.powercalc.nl/img/logo2_light.svg" width="400">
 
-**PowerCalc** is a versatile custom component for Home Assistant that estimates power consumption for devices like lights, fans, smart speakers, and more—especially those without built-in power meters. 
+**PowerCalc** is a versatile custom component for Home Assistant that estimates power consumption for devices like lights, fans, smart speakers, and more—especially those without built-in power meters.
 It acts as a virtual energy monitor, using advanced strategies to calculate power usage. For light entities, PowerCalc analyzes factors such as brightness, hue, saturation, and color temperature to deliver accurate consumption estimates. For other devices, it offers extensive configuration possibilities.
 
 Additionally, PowerCalc includes a powerful measurement utility, enabling users to assess their devices' power usage and contribute custom power profiles to the growing PowerCalc library.

custom_components/powercalc/config_flow.py

@@ -32,6 +32,7 @@
 )
 from homeassistant.core import callback
 from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers import entity_registry as er
 from homeassistant.helpers import selector
 from homeassistant.helpers.schema_config_entry_flow import SchemaFlowError
 from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
@@ -198,15 +199,15 @@ class Step(StrEnum):
 LIBRARY_URL = "https://library.powercalc.nl"
 UNIQUE_ID_TRACKED_UNTRACKED = "pc_tracked_untracked"
 
-STRATEGY_STEP_MAPPING = {
+STRATEGY_STEP_MAPPING: dict[CalculationStrategy, Step] = {
     CalculationStrategy.FIXED: Step.FIXED,
     CalculationStrategy.LINEAR: Step.LINEAR,
     CalculationStrategy.MULTI_SWITCH: Step.MULTI_SWITCH,
     CalculationStrategy.PLAYBOOK: Step.PLAYBOOK,
     CalculationStrategy.WLED: Step.WLED,
 }
 
-GROUP_STEP_MAPPING = {
+GROUP_STEP_MAPPING: dict[GroupType, Step] = {
     GroupType.CUSTOM: Step.GROUP_CUSTOM,
     GroupType.DOMAIN: Step.GROUP_DOMAIN,
     GroupType.STANDBY: Step.GROUP_DOMAIN,
@@ -366,16 +367,8 @@ class Step(StrEnum):
     },
 )
 
-SCHEMA_POWER_MULTI_SWITCH = vol.Schema(
-    {
-        vol.Required(CONF_ENTITIES): selector.EntitySelector(
-            selector.EntitySelectorConfig(domain=Platform.SWITCH, multiple=True),
-        ),
-    },
-)
 SCHEMA_POWER_MULTI_SWITCH_MANUAL = vol.Schema(
     {
-        **SCHEMA_POWER_MULTI_SWITCH.schema,
         vol.Required(CONF_POWER): vol.Coerce(float),
         vol.Required(CONF_POWER_OFF): vol.Coerce(float),
     },
@@ -552,13 +545,19 @@ class Step(StrEnum):
     },
 )
 
-GROUP_SCHEMAS = {
+GROUP_SCHEMAS: dict[GroupType, vol.Schema] = {
     GroupType.CUSTOM: SCHEMA_GROUP,
     GroupType.DOMAIN: SCHEMA_GROUP_DOMAIN,
     GroupType.SUBTRACT: SCHEMA_GROUP_SUBTRACT,
     GroupType.TRACKED_UNTRACKED: SCHEMA_GROUP_TRACKED_UNTRACKED,
 }
 
+STRATEGY_SCHEMAS: dict[CalculationStrategy, vol.Schema] = {
+    CalculationStrategy.FIXED: SCHEMA_POWER_FIXED,
+    CalculationStrategy.PLAYBOOK: SCHEMA_POWER_PLAYBOOK,
+    CalculationStrategy.WLED: SCHEMA_POWER_WLED,
+}
+
 
 @dataclass(slots=True)
 class PowercalcFormStep:
@@ -590,6 +589,7 @@ def __init__(self) -> None:
         self.is_library_flow: bool = False
         self.skip_advanced_step: bool = False
         self.is_options_flow: bool = isinstance(self, OptionsFlow)
+        self.strategy: CalculationStrategy | None = None
         self.name: str | None = None
         super().__init__()
 
@@ -625,26 +625,72 @@ def validate_group_input(user_input: dict[str, Any] | None = None) -> None:
         if not any(key in (user_input or {}) for key in required_keys):
             raise SchemaFlowError("group_mandatory")
 
-    def create_strategy_schema(self, strategy: CalculationStrategy, source_entity_id: str) -> vol.Schema:
+    def create_strategy_schema(self) -> vol.Schema:
         """Get the config schema for a given power calculation strategy."""
-        if strategy == CalculationStrategy.LINEAR:
-            return SCHEMA_POWER_LINEAR.extend(  # type: ignore
+        if not self.strategy:
+            raise ValueError("No strategy selected")  # pragma: no cover
+
+        if hasattr(self, f"create_schema_{self.strategy.value.lower()}"):
+            return getattr(self, f"create_schema_{self.strategy.value.lower()}")()  # type: ignore
+
+        return STRATEGY_SCHEMAS[self.strategy]
+
+    def create_schema_linear(self) -> vol.Schema:
+        """Create the config schema for linear strategy."""
+        return SCHEMA_POWER_LINEAR.extend(  # type: ignore
+            {
+                vol.Optional(CONF_ATTRIBUTE): selector.AttributeSelector(
+                    selector.AttributeSelectorConfig(
+                        entity_id=self.source_entity_id,  # type: ignore
+                        hide_attributes=[],
+                    ),
+                ),
+            },
+        )
+
+    def create_schema_multi_switch(self) -> vol.Schema:
+        """Create the config schema for multi switch strategy."""
+
+        entity_registry = er.async_get(self.hass)
+        entities: list[str] = []
+        if self.source_entity and self.source_entity.device_entry:
+            entities = [
+                entity.entity_id
+                for entity in entity_registry.entities.get_entries_for_device_id(self.source_entity.device_entry.id)
+                if entity.domain == Platform.SWITCH
+            ]
+
+        # pre-populate the entity selector with the switches from the device
+        if entities:
+            schema = vol.Schema(
+                {
+                    vol.Required(
+                        CONF_ENTITIES,
+                        description={"suggested_value": entities},
+                    ): selector.EntitySelector(
+                        selector.EntitySelectorConfig(
+                            domain=Platform.SWITCH,
+                            multiple=True,
+                            include_entities=entities,
+                        ),
+                    ),
+                },
+            )
+        else:
+            schema = vol.Schema(
                 {
-                    vol.Optional(CONF_ATTRIBUTE): selector.AttributeSelector(
-                        selector.AttributeSelectorConfig(
-                            entity_id=source_entity_id,
-                            hide_attributes=[],
+                    vol.Required(CONF_ENTITIES): selector.EntitySelector(
+                        selector.EntitySelectorConfig(
+                            domain=Platform.SWITCH,
+                            multiple=True,
                         ),
                     ),
                 },
             )
-        if strategy == CalculationStrategy.PLAYBOOK:
-            return SCHEMA_POWER_PLAYBOOK
-        if strategy == CalculationStrategy.MULTI_SWITCH:
-            return SCHEMA_POWER_MULTI_SWITCH if self.is_library_flow else SCHEMA_POWER_MULTI_SWITCH_MANUAL
-        if strategy == CalculationStrategy.WLED:
-            return SCHEMA_POWER_WLED
-        return SCHEMA_POWER_FIXED
+        if not self.is_library_flow:
+            schema = schema.extend(SCHEMA_POWER_MULTI_SWITCH_MANUAL.schema)
+
+        return schema
 
     def create_schema_group_custom(
         self,
@@ -772,12 +818,10 @@ def create_group_selector(
 
     def build_strategy_config(
         self,
-        strategy: CalculationStrategy,
-        source_entity_id: str,
         user_input: dict[str, Any],
     ) -> dict[str, Any]:
         """Build the config dict needed for the configured strategy."""
-        strategy_schema = self.create_strategy_schema(strategy, source_entity_id)
+        strategy_schema = self.create_strategy_schema()
         strategy_options: dict[str, Any] = {}
         for key in strategy_schema.schema:
             if user_input.get(key) is None:
@@ -1120,13 +1164,15 @@ async def handle_strategy_step(
         user_input: dict[str, Any] | None = None,
         validate: Callable[[dict[str, Any]], None] | None = None,
     ) -> FlowResult:
+        self.strategy = strategy
+
         async def _validate(user_input: dict[str, Any]) -> dict[str, Any]:
             if validate:
                 validate(user_input)
             await self.validate_strategy_config({strategy: user_input})
             return {strategy: user_input}
 
-        schema = self.create_strategy_schema(strategy, self.source_entity_id)  # type: ignore
+        schema = self.create_strategy_schema()
 
         return await self.handle_form_step(
             PowercalcFormStep(
@@ -1634,7 +1680,7 @@ def __init__(self, config_entry: ConfigEntry) -> None:
         self.sensor_config = dict(config_entry.data)
         self.sensor_type: SensorType = self.sensor_config.get(CONF_SENSOR_TYPE) or SensorType.VIRTUAL_POWER
         self.source_entity_id: str = self.sensor_config.get(CONF_ENTITY_ID)  # type: ignore
-        self.strategy: CalculationStrategy | None = self.sensor_config.get(CONF_MODE)
+        self.strategy = self.sensor_config.get(CONF_MODE)
 
     async def async_step_init(
         self,
@@ -1856,7 +1902,7 @@ async def async_handle_strategy_options_step(self, user_input: dict[str, Any] |
 
         step = STRATEGY_STEP_MAPPING.get(self.strategy, Step.FIXED)
 
-        schema = self.create_strategy_schema(self.strategy, self.source_entity_id)
+        schema = self.create_strategy_schema()
         if self.selected_profile and self.selected_profile.device_type == DeviceType.SMART_SWITCH:
             schema = SCHEMA_POWER_SMART_SWITCH
 
@@ -1905,11 +1951,7 @@ async def process_all_options(self, user_input: dict[str, Any], schema: vol.Sche
                 self._process_user_input(user_input, SCHEMA_POWER_SMART_SWITCH)
                 user_input = {CONF_POWER: user_input.get(CONF_POWER, 0)}
 
-            strategy_options = self.build_strategy_config(
-                self.strategy,
-                self.source_entity_id,
-                user_input or {},
-            )
+            strategy_options = self.build_strategy_config(user_input or {})
 
             if self.strategy != CalculationStrategy.LUT:
                 self.sensor_config.update({str(self.strategy): strategy_options})

custom_components/powercalc/sensors/power.py

@@ -87,6 +87,7 @@
 from custom_components.powercalc.helpers import evaluate_power
 from custom_components.powercalc.power_profile.factory import get_power_profile
 from custom_components.powercalc.power_profile.power_profile import (
+    DiscoveryBy,
     PowerProfile,
     SubProfileSelectConfig,
     SubProfileSelector,
@@ -505,7 +506,8 @@ async def _handle_source_entity_state_change(
             self._sleep_power_timer()
             self._sleep_power_timer = None
 
-        if self.source_entity == DUMMY_ENTITY_ID:
+        discovery_by = self._power_profile.discovery_by if self._power_profile else DiscoveryBy.ENTITY
+        if self.source_entity == DUMMY_ENTITY_ID and discovery_by == DiscoveryBy.ENTITY:
             state = State(self.source_entity, STATE_ON)
 
         if not state or not self._has_valid_state(state):

docs/source/library/device-types/smart-switch.md

@@ -71,6 +71,7 @@ Examples of this type of smart switch are:
 ```json
 {
   "calculation_strategy": "multi_switch",
+  "discovery_by": "device",
   "multi_switch_config": {
     "power": 0.8,
     "power_off": 0.25

profile_library/tp-link/HS300/model.json

@@ -1,6 +1,7 @@
 {
   "name": "Kasa Smart Wi-Fi Power Strip",
   "device_type": "smart_switch",
+  "discovery_by": "device",
   "calculation_strategy": "multi_switch",
   "measure_method": "manual",
   "measure_device": "Kill-a-Watt",
@@ -14,5 +15,6 @@
   "aliases": [
     "HS300(US)"
   ],
+  "only_self_usage": true,
   "author": "Bram Gerritsen <[email protected]>"
 }