Add example with BatteryStatus working alone

Signed-off-by: ela-kotulska-frequenz <[email protected]>

Author: ela-kotulska-frequenz

examples/battery_status.py

@@ -0,0 +1,44 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""Example how to run BatteryPoolStatus as separate instance.
+
+This is not needed for user but simplifies testing and debugging and understanding
+this feature.
+"""
+
+import asyncio
+import logging
+
+from frequenz.sdk import microgrid
+from frequenz.sdk.actor.power_distributing._battery_pool_status import BatteryPoolStatus
+from frequenz.sdk.microgrid.component import ComponentCategory
+
+_logger = logging.getLogger(__name__)
+HOST = "microgrid.sandbox.api.frequenz.io"  # it should be the host name.
+PORT = 61060
+
+
+async def main() -> None:
+    """Start BatteryPoolStatus to see how it works."""
+    logging.basicConfig(
+        level=logging.DEBUG, format="%(asctime)s %(name)s %(levelname)s:%(message)s"
+    )
+    await microgrid.initialize(HOST, PORT)
+    batteries = {
+        bat.component_id
+        for bat in microgrid.get().component_graph.components(
+            component_category={ComponentCategory.BATTERY}
+        )
+    }
+
+    batteries_status = BatteryPoolStatus(
+        battery_ids=batteries,
+        max_data_age_sec=5,
+        max_blocking_duration_sec=30,
+    )
+
+    await batteries_status.join()
+
+
+asyncio.run(main())

src/frequenz/sdk/actor/power_distributing/_battery_pool_status.py

@@ -123,6 +123,17 @@ def __init__(
 
         self._task = asyncio.create_task(self._run())
 
+    async def join(self) -> None:
+        """Await for the battery pool, and return when the task completes.
+
+        It will not terminate the program while BatteryPool is working.
+        BatteryPool can be stopped with the `stop` method.
+        This method is not needed in source code, because BatteryPool is owned
+        by the internal code.
+        It is needed only when user needs to run his own instance of the BatteryPool.
+        """
+        await self._task
+
     async def stop(self) -> None:
         """Stop tracking batteries status."""
         await cancel_and_await(self._task)