Commit WIP

custom_components/powercalc/config_flow.py

@@ -28,6 +28,7 @@
 from .common import SourceEntity, create_source_entity
 from .const import (
     CONF_AREA,
+    CONF_AUTOSTART,
     CONF_CALCULATION_ENABLED_CONDITION,
     CONF_CALIBRATE,
     CONF_CREATE_ENERGY_SENSOR,
@@ -50,8 +51,11 @@
     CONF_MODEL,
     CONF_MULTIPLY_FACTOR,
     CONF_ON_TIME,
+    CONF_PLAYBOOK,
+    CONF_PLAYBOOKS,
     CONF_POWER,
     CONF_POWER_TEMPLATE,
+    CONF_REPEAT,
     CONF_SENSOR_TYPE,
     CONF_STANDBY_POWER,
     CONF_STATES_POWER,
@@ -191,6 +195,7 @@
         options=[
             CalculationStrategy.FIXED,
             CalculationStrategy.LINEAR,
+            CalculationStrategy.PLAYBOOK,
             CalculationStrategy.WLED,
             CalculationStrategy.LUT,
         ],
@@ -215,6 +220,14 @@
     },
 )
 
+SCHEMA_POWER_PLAYBOOK = vol.Schema(
+    {
+        vol.Optional(CONF_PLAYBOOKS): selector.ObjectSelector(),
+        vol.Optional(CONF_REPEAT): selector.BooleanSelector(),
+        vol.Optional(CONF_AUTOSTART): selector.TextSelector(),
+    },
+)
+
 SCHEMA_POWER_AUTODISCOVERED = vol.Schema(
     {vol.Optional(CONF_CONFIRM_AUTODISCOVERED_MODEL, default=True): bool},
 )
@@ -359,6 +372,9 @@ async def async_step_virtual_power(
             if user_input.get(CONF_MODE) == CalculationStrategy.LINEAR:
                 return await self.async_step_linear()
 
+            if user_input.get(CONF_MODE) == CalculationStrategy.PLAYBOOK:
+                return await self.async_step_playbook()
+
             if user_input.get(CONF_MODE) == CalculationStrategy.WLED:
                 return await self.async_step_wled()
 
@@ -455,6 +471,21 @@ async def async_step_linear(
             last_step=False,
         )
 
+    async def async_step_playbook(
+        self,
+        user_input: dict[str, Any] | None = None,
+    ) -> FlowResult:
+        if user_input is not None:
+            self.sensor_config.update({CONF_PLAYBOOK: user_input})
+            return await self.async_step_power_advanced()
+
+        return self.async_show_form(
+            step_id="playbook",
+            data_schema=SCHEMA_POWER_PLAYBOOK,
+            errors={},
+            last_step=False,
+        )
+
     async def async_step_wled(
         self,
         user_input: dict[str, Any] | None = None,
@@ -873,6 +904,8 @@ def _get_strategy_schema(strategy: str, source_entity_id: str) -> vol.Schema:
         return SCHEMA_POWER_FIXED
     if strategy == CalculationStrategy.LINEAR:
         return _create_linear_schema(source_entity_id)
+    if strategy == CalculationStrategy.PLAYBOOK:
+        return SCHEMA_POWER_PLAYBOOK
     if strategy == CalculationStrategy.WLED:
         return SCHEMA_POWER_WLED
     return vol.Schema({})

custom_components/powercalc/strategy/playbook.py

@@ -114,6 +114,13 @@ def _execute_playbook_entry(self) -> None:
 
         queue = self._active_playbook.queue
         if len(queue) == 0:
+            if self._repeat:
+                _LOGGER.debug(f"Playbook {self._active_playbook.key} repeating")
+                self._start_time = dt.utcnow()
+                queue.reset()
+                self._execute_playbook_entry()
+                return
+
             _LOGGER.debug(f"Playbook {self._active_playbook.key} completed")
             self._active_playbook = None
             return
@@ -153,33 +160,32 @@ async def _load_playbook(self, playbook_id: str) -> Playbook:
             )
 
         with open(file_path) as csv_file:
-            queue = PlaybookQueue(repeat=self._repeat)
-
             csv_reader = csv.reader(csv_file)
+            entries = []
             for row in csv_reader:
-                queue.enqueue(PlaybookEntry(time=float(row[0]), power=Decimal(row[1])))
+                entries.append(PlaybookEntry(time=float(row[0]), power=Decimal(row[1])))
 
-            self._loaded_playbooks[playbook_id] = Playbook(key=playbook_id, queue=queue)
+            self._loaded_playbooks[playbook_id] = Playbook(key=playbook_id, queue=PlaybookQueue(entries))
 
         return self._loaded_playbooks[playbook_id]
 
 
 class PlaybookQueue:
-    def __init__(self, repeat: bool = False) -> None:
-        self._entries: deque[PlaybookEntry] = deque()
-        self._repeat = repeat
+    def __init__(self, items: list[PlaybookEntry]) -> None:
+        self._items = items
+        self._queue: deque[PlaybookEntry] = deque(items)
 
     def enqueue(self, entry: PlaybookEntry) -> None:
-        self._entries.append(entry)
+        self._queue.append(entry)
 
     def dequeue(self) -> PlaybookEntry:
-        entry = self._entries.popleft()
-        if self._repeat:
-            self.enqueue(entry)
-        return entry
+        return self._queue.popleft()
+
+    def reset(self) -> None:
+        self._queue = deque(self._items)
 
     def __len__(self) -> int:
-        return len(self._entries)
+        return len(self._queue)
 
 
 @dataclass