steam_boiler_pool

frequenz.sdk.timeseries.steam_boiler_pool

Interactions with pools of steam boilers.

Classes

frequenz.sdk.timeseries.steam_boiler_pool.SteamBoilerPool

Bases: ComponentPool[SteamBoilerPoolReferenceStore, SteamBoilerPoolReport]

An interface for interaction with pools of steam boilers.

Provides

• Aggregate power measurements of the steam boilers in the pool.

Source code in src/frequenz/sdk/timeseries/steam_boiler_pool/_steam_boiler_pool.py

class SteamBoilerPool(
    ComponentPool[SteamBoilerPoolReferenceStore, SteamBoilerPoolReport]
):
    """An interface for interaction with pools of steam boilers.

    Provides:
      - Aggregate
        [`power`][frequenz.sdk.timeseries.steam_boiler_pool.SteamBoilerPool.power]
        measurements of the steam boilers in the pool.
    """

    @override
    async def propose_power(
        self,
        power: Power | None,
        bounds: Bounds[Power | None] = Bounds(None, None),
    ) -> None:
        """Send a proposal to the power manager for the pool's set of steam boilers.

        Steam boilers are controllable loads, so this proposal is for the power the
        boilers in the pool should consume.

        Details on how the power manager handles proposals can be found in the
        [Microgrid][frequenz.sdk.microgrid--setting-power] documentation.

        Args:
            power: The power to propose for the steam boilers in the pool.  If `None`,
                this proposal will not have any effect on the target power, unless
                bounds are specified.  When specified without bounds, bounds for lower
                priority actors will be shifted by this power.  If both are `None`, it
                is equivalent to not having a proposal or withdrawing a previous one.
            bounds: The power bounds for the proposal. When specified, these bounds will
                limit the bounds for lower priority actors.

        Raises:
            SteamBoilerPoolError: If a discharge power for steam boilers is requested.
        """
        if power is not None and power < Power.zero():
            raise SteamBoilerPoolError(
                "Discharging from steam boilers is not supported."
            )
        await super().propose_power(power, bounds=bounds)

    @property
    @override
    def power(self) -> Formula[Power]:
        """Fetch the total power for the steam boilers in the pool.

        This formula produces values that are in the Passive Sign Convention (PSC).

        If a formula to calculate steam boiler power is not already running, it
        will be started.

        A receiver from the formula can be created using the `new_receiver`
        method.

        Returns:
            A Formula that will calculate and stream the total power of all steam boilers.
        """
        return self._pool_ref_store.formula_pool.from_power_formula(
            "steam_boiler_power",
            connection_manager.get().component_graph.steam_boiler_formula(
                self._pool_ref_store.component_ids
            ),
        )

Attributes

component_ids property

component_ids: Set[ComponentId]

Return component IDs of all component IDs managed by this pool.

RETURNS	DESCRIPTION
Set[ComponentId]	Set of managed component IDs.

power property

power: Formula[Power]

Fetch the total power for the steam boilers in the pool.

This formula produces values that are in the Passive Sign Convention (PSC).

If a formula to calculate steam boiler power is not already running, it will be started.

A receiver from the formula can be created using the new_receiver method.

RETURNS	DESCRIPTION
Formula[Power]	A Formula that will calculate and stream the total power of all steam boilers.

power_distribution_results property

power_distribution_results: ReceiverFetcher[Result]

Get a receiver to receive power distribution results.

RETURNS	DESCRIPTION
ReceiverFetcher[Result]	A receiver that will stream power distribution results for the pool's set of
ReceiverFetcher[Result]	components.

power_status property

power_status: ReceiverFetcher[ReportT]

Get a receiver to receive new power status reports when they change.

These include - the current inclusion/exclusion bounds available for the pool's priority, - the current target power for the pool's set of components, - the result of the last distribution request for the pool's set of components,.

RETURNS	DESCRIPTION
ReceiverFetcher[ReportT]	A receiver that will stream power status reports for the pool's priority.

system_power_bounds property

system_power_bounds: ReceiverFetcher[SystemBounds]

Return a receiver fetcher for the system power bounds.

Methods:

__init__

__init__(
    *,
    pool_ref_store: RefStoreT,
    name: str | None,
    priority: int
) -> None

Create an AbstractPool instance.

PARAMETER	DESCRIPTION
pool_ref_store	The pool reference store instance. TYPE: RefStoreT
name	An optional name used to identify this instance of the pool or a corresponding actor in the logs. TYPE: str | None
priority	The priority of the actor using this wrapper. TYPE: int

Source code in src/frequenz/sdk/timeseries/component_pool/_component_pool.py

def __init__(  # pylint: disable=too-many-arguments
    self,
    *,
    pool_ref_store: RefStoreT,
    name: str | None,
    priority: int,
) -> None:
    """Create an `AbstractPool` instance.

    Args:
        pool_ref_store: The pool reference store instance.
        name: An optional name used to identify this instance of the pool or a
            corresponding actor in the logs.
        priority: The priority of the actor using this wrapper.
    """
    self._pool_ref_store = pool_ref_store
    unique_id = str(uuid.uuid4())
    self._source_id = unique_id if name is None else f"{name}-{unique_id}"
    self._priority = priority

propose_power async

propose_power(
    power: Power | None,
    bounds: Bounds[Power | None] = Bounds(None, None),
) -> None

Send a proposal to the power manager for the pool's set of steam boilers.

Steam boilers are controllable loads, so this proposal is for the power the boilers in the pool should consume.

Details on how the power manager handles proposals can be found in the Microgrid documentation.

PARAMETER	DESCRIPTION
power	The power to propose for the steam boilers in the pool. If None, this proposal will not have any effect on the target power, unless bounds are specified. When specified without bounds, bounds for lower priority actors will be shifted by this power. If both are None, it is equivalent to not having a proposal or withdrawing a previous one. TYPE: Power | None
bounds	The power bounds for the proposal. When specified, these bounds will limit the bounds for lower priority actors. TYPE: Bounds[Power | None] DEFAULT: Bounds(None, None)

RAISES	DESCRIPTION
SteamBoilerPoolError	If a discharge power for steam boilers is requested.

Source code in src/frequenz/sdk/timeseries/steam_boiler_pool/_steam_boiler_pool.py

@override
async def propose_power(
    self,
    power: Power | None,
    bounds: Bounds[Power | None] = Bounds(None, None),
) -> None:
    """Send a proposal to the power manager for the pool's set of steam boilers.

    Steam boilers are controllable loads, so this proposal is for the power the
    boilers in the pool should consume.

    Details on how the power manager handles proposals can be found in the
    [Microgrid][frequenz.sdk.microgrid--setting-power] documentation.

    Args:
        power: The power to propose for the steam boilers in the pool.  If `None`,
            this proposal will not have any effect on the target power, unless
            bounds are specified.  When specified without bounds, bounds for lower
            priority actors will be shifted by this power.  If both are `None`, it
            is equivalent to not having a proposal or withdrawing a previous one.
        bounds: The power bounds for the proposal. When specified, these bounds will
            limit the bounds for lower priority actors.

    Raises:
        SteamBoilerPoolError: If a discharge power for steam boilers is requested.
    """
    if power is not None and power < Power.zero():
        raise SteamBoilerPoolError(
            "Discharging from steam boilers is not supported."
        )
    await super().propose_power(power, bounds=bounds)

stop async

stop() -> None

Stop all tasks and channels owned by the pool.

Source code in src/frequenz/sdk/timeseries/component_pool/_component_pool.py

async def stop(self) -> None:
    """Stop all tasks and channels owned by the pool."""

frequenz.sdk.timeseries.steam_boiler_pool.SteamBoilerPoolError

Bases: Exception

An error that occurred in any of the SteamBoilerPool methods.

Source code in src/frequenz/sdk/timeseries/steam_boiler_pool/_steam_boiler_pool.py

class SteamBoilerPoolError(Exception):
    """An error that occurred in any of the SteamBoilerPool methods."""

frequenz.sdk.timeseries.steam_boiler_pool.SteamBoilerPoolReport

Bases: ComponentPoolReport, Protocol

A status report for a steam boiler pool.

Source code in src/frequenz/sdk/timeseries/steam_boiler_pool/_result_types.py

class SteamBoilerPoolReport(ComponentPoolReport, typing.Protocol):
    """A status report for a steam boiler pool."""

    @property
    def target_power(self) -> Power | None:
        """The currently set power for the steam boilers."""

    @property
    def bounds(self) -> Bounds[Power] | None:
        """The usable bounds for the steam boilers.

        These bounds are adjusted to any restrictions placed by actors with higher
        priorities.
        """

Attributes

bounds property

bounds: Bounds[Power] | None

The usable bounds for the steam boilers.

These bounds are adjusted to any restrictions placed by actors with higher priorities.

target_power property

target_power: Power | None

The currently set power for the steam boilers.