2424
2525
2626class DispatchActorId (BaseId , str_prefix = "DA" ):
27- """ID for a dispatch actor."""
27+ """An ID for a dispatch actor."""
2828
2929 def __init__ (self , dispatch_id : DispatchId | int ) -> None :
3030 """Initialize the DispatchActorId.
@@ -37,26 +37,26 @@ def __init__(self, dispatch_id: DispatchId | int) -> None:
3737
3838@dataclass (frozen = True , kw_only = True )
3939class DispatchInfo :
40- """Event emitted when the dispatch changes."""
40+ """An event emitted when the dispatch changes."""
4141
4242 @property
4343 @deprecated ("'components' is deprecated, use 'target' instead." )
4444 def components (self ) -> TargetComponents :
45- """Get the target components.
45+ """The target components.
4646
47- Deprecation : Deprecated in v0.10.3
48- Use [`target`][frequenz.dispatch.DispatchInfo .target] instead.
47+ Warning : Deprecated in v0.10.3
48+ Use [`target`][. .target] instead.
4949 """
5050 return self .target
5151
5252 target : TargetComponents
53- """Target components to be used ."""
53+ """The target components ."""
5454
5555 dry_run : bool
5656 """Whether this is a dry run."""
5757
5858 options : dict [str , Any ]
59- """Additional options."""
59+ """The additional options."""
6060
6161 _src : Dispatch
6262 """The dispatch that triggered this update."""
@@ -70,13 +70,13 @@ def __init__(
7070 options : dict [str , Any ],
7171 _src : Dispatch ,
7272 ) -> None :
73- """Initialize the DispatchInfo .
73+ """Initialize a new instance .
7474
7575 Args:
76- target: Target components to be used.
76+ target: The target components to be used.
7777 components: Deprecated alias for `target`.
7878 dry_run: Whether this is a dry run.
79- options: Additional options.
79+ options: The additional options.
8080 _src: The dispatch that triggered this update.
8181
8282 Raises:
@@ -103,109 +103,109 @@ def __init__(
103103
104104
105105class ActorDispatcher (BackgroundService ):
106- """Helper class to manage actors based on dispatches.
107-
108- Example usage:
109-
110- ```python
111- import os
112- import asyncio
113- from typing import override
114- from frequenz.dispatch import Dispatcher, ActorDispatcher, DispatchInfo
115- from frequenz.client.common.microgrid.components import ComponentCategory
116- from frequenz.channels import Receiver, Broadcast, select, selected_from
117- from frequenz.sdk.actor import Actor, run
118-
119- class MyActor(Actor):
120- def __init__(
121- self,
122- *,
123- name: str | None = None,
124- ) -> None:
125- super().__init__(name=name)
126- self._dispatch_updates_receiver: Receiver[DispatchInfo] | None = None
127- self._dry_run: bool = False
128- self._options: dict[str, Any] = {}
129-
130- @classmethod
131- def new_with_dispatch(
132- cls,
133- initial_dispatch: DispatchInfo,
134- dispatch_updates_receiver: Receiver[DispatchInfo],
135- *,
136- name: str | None = None,
137- ) -> "Self":
138- self = cls(name=name)
139- self._dispatch_updates_receiver = dispatch_updates_receiver
140- self._update_dispatch_information(initial_dispatch)
141- return self
142-
143- @override
144- async def _run(self) -> None:
145- other_recv: Receiver[Any] = ...
146-
147- if self._dispatch_updates_receiver is None:
148- async for msg in other_recv:
149- # do stuff
150- ...
151- else:
152- await self._run_with_dispatch(other_recv)
153-
154- async def _run_with_dispatch(self, other_recv: Receiver[Any]) -> None:
155- async for selected in select(self._dispatch_updates_receiver, other_recv):
156- if selected_from(selected, self._dispatch_updates_receiver):
157- self._update_dispatch_information(selected.message)
158- elif selected_from(selected, other_recv):
159- # do stuff
160- ...
106+ """A helper class to manage actors based on dispatches.
107+
108+ Example:
109+ ```python
110+ import os
111+ import asyncio
112+ from typing import Any, Self
113+ from typing import override
114+ from frequenz.dispatch import Dispatcher, ActorDispatcher, DispatchInfo
115+ from frequenz.client.common.microgrid.components import ComponentCategory
116+ from frequenz.channels import Receiver, Broadcast, select, selected_from
117+ from frequenz.sdk.actor import Actor, run
118+
119+ class MyActor(Actor):
120+ def __init__(
121+ self,
122+ *,
123+ name: str | None = None,
124+ ) -> None:
125+ super().__init__(name=name)
126+ self._dispatch_updates_receiver: Receiver[DispatchInfo] | None = None
127+ self._dry_run: bool = False
128+ self._options: dict[str, Any] = {}
129+
130+ @classmethod
131+ def new_with_dispatch(
132+ cls,
133+ initial_dispatch: DispatchInfo,
134+ dispatch_updates_receiver: Receiver[DispatchInfo],
135+ *,
136+ name: str | None = None,
137+ ) -> Self:
138+ self = cls(name=name)
139+ self._dispatch_updates_receiver = dispatch_updates_receiver
140+ self._update_dispatch_information(initial_dispatch)
141+ return self
142+
143+ @override
144+ async def _run(self) -> None:
145+ other_recv: Receiver[Any] = ...
146+
147+ if self._dispatch_updates_receiver is None:
148+ async for msg in other_recv:
149+ # do stuff
150+ ...
161151 else:
162- assert False, f"Unexpected selected receiver: {selected}"
163-
164- def _update_dispatch_information(self, dispatch_update: DispatchInfo) -> None:
165- print("Received update:", dispatch_update)
166- self._dry_run = dispatch_update.dry_run
167- self._options = dispatch_update.options
168- match dispatch_update.components:
169- case []:
170- print("Dispatch: Using all components")
171- case list() as ids if isinstance(ids[0], int):
172- component_ids = ids
173- case [ComponentCategory.BATTERY, *_]:
174- component_category = ComponentCategory.BATTERY
175- case unsupported:
176- print(
177- "Dispatch: Requested an unsupported selector %r, "
178- "but only component IDs or category BATTERY are supported.",
179- unsupported,
180- )
181-
182- async def main():
183- url = os.getenv("DISPATCH_API_URL", "grpc://dispatch.url.goes.here.example.com")
184- auth_key = os.getenv("DISPATCH_API_AUTH_KEY", "some-key")
185- sign_secret = os.getenv("DISPATCH_API_SIGN_SECRET")
186-
187- microgrid_id = 1
188-
189- async with Dispatcher(
190- microgrid_id=microgrid_id,
191- server_url=url,
192- auth_key=auth_key,
193- sign_secret=sign_secret,
194- ) as dispatcher:
195- status_receiver = dispatcher.new_running_state_event_receiver("EXAMPLE_TYPE")
196-
197- managing_actor = ActorDispatcher(
198- actor_factory=MyActor.new_with_dispatch,
199- running_status_receiver=status_receiver,
200- )
152+ await self._run_with_dispatch(other_recv)
153+
154+ async def _run_with_dispatch(self, other_recv: Receiver[Any]) -> None:
155+ async for selected in select(self._dispatch_updates_receiver, other_recv):
156+ if selected_from(selected, self._dispatch_updates_receiver):
157+ self._update_dispatch_information(selected.message)
158+ elif selected_from(selected, other_recv):
159+ # do stuff
160+ ...
161+ else:
162+ assert False, f"Unexpected selected receiver: {selected}"
163+
164+ def _update_dispatch_information(self, dispatch_update: DispatchInfo) -> None:
165+ print("Received update:", dispatch_update)
166+ self._dry_run = dispatch_update.dry_run
167+ self._options = dispatch_update.options
168+ match dispatch_update.target:
169+ case []:
170+ print("Dispatch: Using all components")
171+ case list() as ids if isinstance(ids[0], int):
172+ component_ids = ids
173+ case [ComponentCategory.BATTERY, *_]:
174+ component_category = ComponentCategory.BATTERY
175+ case unsupported:
176+ print(
177+ "Dispatch: Requested an unsupported selector %r, "
178+ "but only component IDs or category BATTERY are supported.",
179+ unsupported,
180+ )
181+
182+ async def main():
183+ url = os.getenv("DISPATCH_API_URL", "grpc://dispatch.url.goes.here.example.com")
184+ auth_key = os.getenv("DISPATCH_API_AUTH_KEY", "some-key")
185+ sign_secret = os.getenv("DISPATCH_API_SIGN_SECRET")
186+
187+ microgrid_id = 1
188+
189+ async with Dispatcher(
190+ microgrid_id=microgrid_id,
191+ server_url=url,
192+ auth_key=auth_key,
193+ sign_secret=sign_secret,
194+ ) as dispatcher:
195+ status_receiver = await dispatcher.new_running_state_event_receiver("EXAMPLE_TYPE")
196+
197+ managing_actor = ActorDispatcher(
198+ actor_factory=MyActor.new_with_dispatch,
199+ running_status_receiver=status_receiver,
200+ )
201201
202- await run(managing_actor)
203- ```
202+ await run(managing_actor)
203+ ```
204204 """
205205
206206 @dataclass (frozen = True , kw_only = True )
207207 class ActorAndChannel :
208- """Actor and its sender."""
208+ """An actor and its dispatch update sender."""
209209
210210 actor : Actor
211211 """The actor."""
@@ -252,7 +252,11 @@ def start(self) -> None:
252252 self ._tasks .add (asyncio .create_task (self ._run ()))
253253
254254 async def _start_actor (self , dispatch : Dispatch ) -> None :
255- """Start the actor the given dispatch refers to."""
255+ """Start the actor the given dispatch refers to.
256+
257+ Args:
258+ dispatch: The dispatch to start the actor for.
259+ """
256260 dispatch_update = DispatchInfo (
257261 target = dispatch .target ,
258262 dry_run = dispatch .dry_run ,
@@ -298,7 +302,7 @@ async def _start_actor(self, dispatch: Dispatch) -> None:
298302 )
299303
300304 async def _stop_actor (self , stopping_dispatch : Dispatch , msg : str ) -> None :
301- """Stop all actors .
305+ """Stop the actor for the given dispatch .
302306
303307 Args:
304308 stopping_dispatch: The dispatch that is stopping the actor.
0 commit comments