11# License: MIT
22# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
33
4- """Actor to distribute power between batteries .
4+ """Actor to distribute power between components .
55
6- When charge/discharge method is called the power should be distributed so that
7- the SoC in batteries stays at the same level. That way of distribution
8- prevents using only one battery, increasing temperature, and maximize the total
9- amount power to charge/discharge.
6+ The purpose of this actor is to distribute power between components in a microgrid.
107
11- Purpose of this actor is to keep SoC level of each component at the equal level.
8+ The actor receives power requests from the power manager, process them by
9+ distributing the power between the components and sends the results back to it.
1210"""
1311
1412
13+ import asyncio
14+ import logging
1515from datetime import timedelta
1616
1717from frequenz .channels import Receiver , Sender
2929from .request import Request
3030from .result import Result
3131
32+ _logger = logging .getLogger (__name__ )
33+
3234
3335class PowerDistributingActor (Actor ):
3436 # pylint: disable=too-many-instance-attributes
35- """Actor to distribute the power between batteries in a microgrid.
36-
37- The purpose of this tool is to keep an equal SoC level in all batteries.
38- The PowerDistributingActor can have many concurrent users which at this time
39- need to be known at construction time.
37+ """Actor to distribute the power between components in a microgrid.
4038
41- For each user a bidirectional channel needs to be created through which
42- they can send and receive requests and responses.
39+ One instance of the actor can handle only one component category and type,
40+ which needs to be specified at actor startup and it will setup the correct
41+ component manager based on the given category and type.
4342
44- It is recommended to wait for PowerDistributingActor output with timeout. Otherwise if
45- the processing function fails then the response will never come.
46- The timeout should be Result:request_timeout + time for processing the request .
43+ Only one power request is processed at a time to prevent from sending
44+ multiple requests for the same components to the microgrid API at the
45+ same time.
4746
4847 Edge cases:
49- * If there are 2 requests to be processed for the same subset of batteries, then
50- only the latest request will be processed. Older request will be ignored. User with
51- older request will get response with Result.Status.IGNORED.
52-
53- * If there are 2 requests and their subset of batteries is different but they
54- overlap (they have at least one common battery), then then both batteries
55- will be processed. However it is not expected so the proper error log will be
56- printed.
48+ * If a new power request is received while a power request with the same
49+ set of components is being processed, the new request will be added to
50+ the pending requests. Then the pending request will be processed after the
51+ request with the same set of components being processed is done. Only one
52+ pending request is kept for each set of components, the latest request will
53+ overwrite the previous one if there is any.
54+
55+ * If there are 2 requests and their set of components is different but they
56+ overlap (they have at least one common component), then both requests will
57+ be processed concurrently. Though, the power manager will make sure this
58+ doesn't happen as overlapping component IDs are not possible at the moment.
5759 """
5860
5961 def __init__ ( # pylint: disable=too-many-arguments
@@ -67,7 +69,7 @@ def __init__( # pylint: disable=too-many-arguments
6769 component_type : ComponentType | None = None ,
6870 name : str | None = None ,
6971 ) -> None :
70- """Create class instance.
72+ """Create actor instance.
7173
7274 Args:
7375 requests_receiver: Receiver for receiving power requests from the power
@@ -99,6 +101,16 @@ def __init__( # pylint: disable=too-many-arguments
99101 self ._result_sender = results_sender
100102 self ._api_power_request_timeout = api_power_request_timeout
101103
104+ self ._processing_tasks : dict [frozenset [int ], asyncio .Task [None ]] = {}
105+ """Track the power request tasks currently being processed."""
106+
107+ self ._pending_requests : dict [frozenset [int ], Request ] = {}
108+ """Track the power requests that are waiting to be processed.
109+
110+ Only one pending power request is kept for each set of components, the
111+ latest request will overwrite the previous one.
112+ """
113+
102114 self ._component_manager : ComponentManager
103115 if component_category == ComponentCategory .BATTERY :
104116 self ._component_manager = BatteryManager (
@@ -121,19 +133,34 @@ def __init__( # pylint: disable=too-many-arguments
121133 )
122134
123135 @override
124- async def _run (self ) -> None : # pylint: disable=too-many-locals
125- """Run actor main function.
136+ async def _run (self ) -> None :
137+ """Run this actor's logic.
138+
139+ It waits for new power requests and process them. Only one power request
140+ can be processed at a time to prevent from sending multiple requests for
141+ the same components to the microgrid API at the same time.
126142
127- It waits for new requests in task_queue and process it, and send
128- `set_power` request with distributed power .
129- The output of the `set_power` method is processed.
130- Every battery and inverter that failed or didn't respond in time will be marked
143+ A new power request will be ignored if a power request with the same
144+ components is currently being processed .
145+
146+ Every component that failed or didn't respond in time will be marked
131147 as broken for some time.
132148 """
133149 await self ._component_manager .start ()
134150
135151 async for request in self ._requests_receiver :
136- await self ._component_manager .distribute_power (request )
152+ req_id = frozenset (request .component_ids )
153+
154+ if req_id in self ._processing_tasks :
155+ if pending_request := self ._pending_requests .get (req_id ):
156+ _logger .debug (
157+ "Pending request: %s, overwritten with request: %s" ,
158+ pending_request ,
159+ request ,
160+ )
161+ self ._pending_requests [req_id ] = request
162+ else :
163+ self ._process_request (req_id , request )
137164
138165 @override
139166 async def stop (self , msg : str | None = None ) -> None :
@@ -144,3 +171,41 @@ async def stop(self, msg: str | None = None) -> None:
144171 """
145172 await self ._component_manager .stop ()
146173 await super ().stop (msg )
174+
175+ def _handle_task_completion (
176+ self , req_id : frozenset [int ], request : Request , task : asyncio .Task [None ]
177+ ) -> None :
178+ """Handle the completion of a power request task.
179+
180+ Args:
181+ req_id: The id to identify the power request.
182+ request: The power request that has been processed.
183+ task: The task that has completed.
184+ """
185+ try :
186+ task .result ()
187+ except Exception : # pylint: disable=broad-except
188+ _logger .exception ("Failed power request: %s" , request )
189+
190+ if req_id in self ._pending_requests :
191+ self ._process_request (req_id , self ._pending_requests .pop (req_id ))
192+ elif req_id in self ._processing_tasks :
193+ del self ._processing_tasks [req_id ]
194+ else :
195+ _logger .error ("Request id not found in processing tasks: %s" , req_id )
196+
197+ def _process_request (self , req_id : frozenset [int ], request : Request ) -> None :
198+ """Process a power request.
199+
200+ Args:
201+ req_id: The id to identify the power request.
202+ request: The power request to process.
203+ """
204+ task = asyncio .create_task (
205+ self ._component_manager .distribute_power (request ),
206+ name = f"{ type (self ).__name__ } { request } ,
207+ )
208+ task .add_done_callback (
209+ lambda t : self ._handle_task_completion (req_id , request , t )
210+ )
211+ self ._processing_tasks [req_id ] = task
0 commit comments