Add load_shedding example

Signed-off-by: Mathias L. Baumann <mathias.baumann@frequenz.com>

examples/load_shedding.py

@@ -0,0 +1,243 @@
+import asyncio
+import random
+import sys
+import termios
+import tty
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from heapq import heappop, heappush
+from unittest.mock import patch
+
+from frequenz.quantities import Percentage, Power
+
+from frequenz.sdk import microgrid
+from frequenz.sdk.actor import Actor, run
+from frequenz.sdk.timeseries import Sample
+
+# Mock configuration
+CONF_STATE = {}
+
+
+def mock_set_consumer(name: str, power: float):
+    """
+    Mock setting consumer power by storing the state in a dictionary.
+
+    Args:
+        name (str): Consumer name.
+        power (float): Power value to set.
+    """
+    CONF_STATE[name] = power
+
+
+def log(msg: str):
+    print(msg, end="\n\r")
+
+
+class PowerMockActor(Actor):
+    """
+    Power Mock Actor
+
+    Asynchronously listens to user key presses 'm' and 'n' to increase and decrease power of a
+    static consumer.
+    """
+
+    def __init__(self, consumer_name: str):
+        super().__init__()
+        self.consumer_name = consumer_name
+        self.power_step = Power.from_kilowatts(1.0)
+
+    async def _run(self):
+        log("Press 'm' to increase power or 'n' to decrease power for the consumer.")
+
+        loop = asyncio.get_running_loop()
+        reader = asyncio.StreamReader()
+        protocol = asyncio.StreamReaderProtocol(reader)
+        await loop.connect_read_pipe(lambda: protocol, sys.stdin)
+
+        while True:
+            key = (await reader.read(1)).decode("utf-8")
+            if key == "m":
+                CONF_STATE[self.consumer_name] = (
+                    CONF_STATE.get(self.consumer_name, 0) + self.power_step.as_watts()
+                )
+                log(
+                    f"Increased {self.consumer_name} power to {CONF_STATE[self.consumer_name]} W"
+                )
+            elif key == "n":
+                CONF_STATE[self.consumer_name] = max(
+                    0,
+                    CONF_STATE.get(self.consumer_name, 0) - self.power_step.as_watts(),
+                )
+                log(
+                    f"Decreased {self.consumer_name} power to {CONF_STATE[self.consumer_name]} W"
+                )
+            else:
+                log("Invalid key. Use 'm' or 'n'.")
+
+    async def _run(self):
+        log("Press 'm' to increase power or 'n' to decrease power for the consumer.")
+
+        while True:
+            # Call _read_key in a thread to avoid blocking the event loop
+            key = await asyncio.to_thread(self._read_key)
+            if key == "m":
+                CONF_STATE[self.consumer_name] = (
+                    CONF_STATE.get(self.consumer_name, 0) + self.power_step.as_watts()
+                )
+                log(
+                    f"Increased {self.consumer_name} power to {CONF_STATE[self.consumer_name]/1000.0} kW"
+                )
+            elif key == "n":
+                CONF_STATE[self.consumer_name] = max(
+                    0,
+                    CONF_STATE.get(self.consumer_name, 0) - self.power_step.as_watts(),
+                )
+                log(
+                    f"Decreased {self.consumer_name} power to {CONF_STATE[self.consumer_name]/1000.0} kW"
+                )
+            elif key == "q":
+                sys.exit()
+            else:
+                log("Invalid key. Use 'm' or 'n'.")
+
+    def _read_key(self):
+        """Read a single key press without waiting for Enter."""
+        fd = sys.stdin.fileno()
+        old_settings = termios.tcgetattr(fd)
+        try:
+            tty.setraw(fd)
+            key = sys.stdin.read(1)
+        finally:
+            termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
+        return key
+
+
+@dataclass(order=True)
+class Consumer:
+    priority: int
+    name: str
+    power: Power
+    enabled: bool = False
+
+
+class LoadSheddingActor(Actor):
+    def __init__(
+        self,
+        max_peak: Power,
+        consumers: list[Consumer],
+        grid_meter_receiver,
+    ):
+        super().__init__()
+        self.max_peak = max_peak
+        self.disable_tolerance = self.max_peak * 0.9
+        self.enable_tolerance = self.max_peak * 0.8
+        self.grid_meter_receiver = grid_meter_receiver
+
+        self.enabled_consumers = []
+        self.disabled_consumers = []
+
+        for c in consumers:
+            heappush(self.disabled_consumers, c)
+
+    async def _enable_consumer(self, consumer: Consumer):
+        if not consumer.enabled:
+            consumer.enabled = True
+            heappush(self.enabled_consumers, consumer)
+            log(f"+++{consumer.name}, +{consumer.power}")
+            mock_set_consumer(consumer.name, consumer.power.as_watts())
+
+    async def _disable_consumer(self, consumer: Consumer):
+        if consumer.enabled:
+            consumer.enabled = False
+            heappush(self.disabled_consumers, consumer)
+            log(f"---{consumer.name}, -{consumer.power}")
+            mock_set_consumer(consumer.name, 0)
+
+    async def _adjust_loads(self, current_load: Power):
+        while current_load > self.disable_tolerance and self.enabled_consumers:
+            c: Consumer = heappop(self.enabled_consumers)
+            await self._disable_consumer(c)
+            current_load -= c.power
+
+        temp_disabled: list[Consumer] = []
+        while self.disabled_consumers:
+            c: Consumer = heappop(self.disabled_consumers)
+            if current_load + c.power <= self.enable_tolerance:
+                await self._enable_consumer(c)
+                current_load += c.power
+            else:
+                heappush(temp_disabled, c)
+                break
+
+        while temp_disabled:
+            heappush(self.disabled_consumers, heappop(temp_disabled))
+
+    async def _run(self) -> None:
+        async for power_sample in self.grid_meter_receiver:
+            if power_sample.value:
+                log(
+                    f"Power: {power_sample.value}, "
+                    f"Peak: {self.max_peak} ({self.disable_tolerance} / {self.enable_tolerance})"
+                    f", Enabled: {', '.join(c.name for c in self.enabled_consumers)}\r"
+                )
+                await self._adjust_loads(power_sample.value)
+
+
+async def mock_receiver():
+    """
+    Mock implementation of a grid meter receiver that sends power values every second.
+    """
+    current_load = Power.from_kilowatts(0.0)
+
+    async def compute_power():
+        """Compute current grid power based on mock state."""
+        return Power.from_watts(sum(CONF_STATE.values()))
+
+    while True:
+        current_load = await compute_power()
+        # Add +- 8% noise to the current load
+        current_load += current_load * Percentage.from_fraction(
+            random.uniform(-0.08, 0.08)
+        )
+        yield Sample(timestamp=datetime.now(tz=timezone.utc), value=current_load)
+        await asyncio.sleep(1)
+
+
+async def main() -> None:
+    consumers = [
+        Consumer(priority=1, name="Fan2", power=Power.from_kilowatts(2.5)),
+        Consumer(priority=2, name="Drier1", power=Power.from_kilowatts(3.0)),
+        Consumer(priority=2, name="Drier2", power=Power.from_kilowatts(2.0)),
+        Consumer(priority=3, name="Conveyor1", power=Power.from_kilowatts(1.5)),
+        Consumer(priority=3, name="Conveyor2", power=Power.from_kilowatts(1.0)),
+        Consumer(priority=4, name="Auger", power=Power.from_kilowatts(2.0)),
+        Consumer(priority=4, name="HopperMixer", power=Power.from_kilowatts(2.5)),
+        Consumer(priority=5, name="SiloVentilation", power=Power.from_kilowatts(1.0)),
+        Consumer(priority=5, name="LoaderArm", power=Power.from_kilowatts(3.0)),
+        Consumer(priority=6, name="SeedCleaner", power=Power.from_kilowatts(2.5)),
+        Consumer(priority=6, name="Sprayer", power=Power.from_kilowatts(2.0)),
+        Consumer(priority=7, name="Grinder", power=Power.from_kilowatts(3.0)),
+        Consumer(priority=7, name="Shaker", power=Power.from_kilowatts(1.5)),
+        Consumer(priority=8, name="Sorter", power=Power.from_kilowatts(2.0)),
+    ]
+
+    for consumer in consumers:
+        mock_set_consumer(consumer.name, 0)
+
+    grid_meter_receiver = microgrid.grid().power.new_receiver()
+
+    actor_instance = LoadSheddingActor(
+        max_peak=Power.from_kilowatts(30),
+        consumers=consumers,
+        grid_meter_receiver=grid_meter_receiver,
+    )
+
+    user_input_actor = PowerMockActor(consumer_name="static_consumer")
+
+    await run(actor_instance, user_input_actor)
+
+
+if __name__ == "__main__":
+    with patch("frequenz.sdk.microgrid.grid") as mock_grid:
+        mock_grid.return_value.power.new_receiver = mock_receiver
+        asyncio.run(main())