Skip to content

pv_pool

frequenz.sdk.timeseries.pv_pool ¤

Interactions with PV inverters.

Classes¤

frequenz.sdk.timeseries.pv_pool.PVPool ¤

An interface for interaction with pools of PV inverters.

Provides
  • Aggregate power measurements of the PV inverters in the pool.
Source code in frequenz/sdk/timeseries/pv_pool/_pv_pool.py
class PVPool:
    """An interface for interaction with pools of PV inverters.

    Provides:
      - Aggregate [`power`][frequenz.sdk.timeseries.pv_pool.PVPool.power]
        measurements of the PV inverters in the pool.
    """

    def __init__(  # pylint: disable=too-many-arguments
        self,
        pv_pool_ref: PVPoolReferenceStore,
        name: str | None,
        priority: int,
    ) -> None:
        """Initialize the instance.

        !!! note
            `PVPool` instances are not meant to be created directly by users. Use the
            [`microgrid.pv_pool`][frequenz.sdk.microgrid.pv_pool] method for creating
            `PVPool` instances.

        Args:
            pv_pool_ref: The reference store for the PV pool.
            name: The name of the PV pool.
            priority: The priority of the PV pool.
        """
        self._pv_pool_ref = pv_pool_ref
        unique_id = uuid.uuid4()
        self._source_id = str(unique_id) if name is None else f"{name}-{unique_id}"
        self._priority = priority

    async def propose_power(
        self,
        power: Power | None,
        *,
        request_timeout: timedelta = timedelta(seconds=5.0),
        bounds: Bounds[Power | None] = Bounds(None, None),
    ) -> None:
        """Send a proposal to the power manager for the pool's set of PV inverters.

        This proposal is for the maximum power that can be set for the PV inverters in
        the pool.  The actual production might be lower.

        Power values need to follow the Passive Sign Convention (PSC). That is, positive
        values indicate charge power and negative values indicate discharge power.
        Only discharge powers are allowed for PV inverters.

        If the same PV inverters are shared by multiple actors, the power manager will
        consider the priority of the actors, the bounds they set, and their preferred
        power, when calculating the target power for the PV inverters.

        The preferred power of lower priority actors will take precedence as long as
        they respect the bounds set by higher priority actors.  If lower priority actors
        request power values outside of the bounds set by higher priority actors, the
        target power will be the closest value to the preferred power that is within the
        bounds.

        When there are no other actors trying to use the same PV inverters, the actor's
        preferred power would be set as the target power, as long as it falls within the
        system power bounds for the PV inverters.

        The result of the request can be accessed using the receiver returned from the
        [`power_status`][frequenz.sdk.timeseries.pv_pool.PVPool.power_status]
        method, which also streams the bounds that an actor should comply with, based on
        its priority.

        Args:
            power: The power to propose for the PV inverters in the pool.  If `None`,
                this proposal will not have any effect on the target power, unless
                bounds are specified.  If both are `None`, it is equivalent to not
                having a proposal or withdrawing a previous one.
            request_timeout: The timeout for the request.
            bounds: The power bounds for the proposal.  These bounds will apply to
                actors with a lower priority, and can be overridden by bounds from
                actors with a higher priority.  If None, the power bounds will be set to
                the maximum power of the batteries in the pool.  This is currently and
                experimental feature.

        Raises:
            PVPoolError: If a charge power for PV inverters is requested.
        """
        if power is not None and power > Power.zero():
            raise PVPoolError("Charge powers for PV inverters is not supported.")
        await self._pv_pool_ref.power_manager_requests_sender.send(
            _power_managing.Proposal(
                source_id=self._source_id,
                preferred_power=power,
                bounds=bounds,
                component_ids=self._pv_pool_ref.component_ids,
                priority=self._priority,
                creation_time=asyncio.get_running_loop().time(),
                request_timeout=request_timeout,
            )
        )

    @property
    def component_ids(self) -> abc.Set[int]:
        """Return component IDs of all PV inverters managed by this PVPool.

        Returns:
            Set of managed component IDs.
        """
        return self._pv_pool_ref.component_ids

    @property
    def power(self) -> FormulaEngine[Power]:
        """Fetch the total power for the PV Inverters in the pool.

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

        If a formula engine to calculate PV Inverter power is not already running, it
        will be started.

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

        Returns:
            A FormulaEngine that will calculate and stream the total power of all PV
                Inverters.
        """
        engine = self._pv_pool_ref.formula_pool.from_power_formula_generator(
            "pv_power",
            PVPowerFormula,
            FormulaGeneratorConfig(
                component_ids=self._pv_pool_ref.component_ids,
            ),
        )
        assert isinstance(engine, FormulaEngine)
        return engine

    @property
    def power_status(self) -> ReceiverFetcher[PVPoolReport]:
        """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 batteries,
          - the result of the last distribution request for the pool's set of batteries.

        Returns:
            A receiver that will stream power status reports for the pool's priority.
        """
        sub = _power_managing.ReportRequest(
            source_id=self._source_id,
            priority=self._priority,
            component_ids=self._pv_pool_ref.component_ids,
        )
        self._pv_pool_ref.power_bounds_subs[sub.get_channel_name()] = (
            asyncio.create_task(
                self._pv_pool_ref.power_manager_bounds_subs_sender.send(sub)
            )
        )
        channel = self._pv_pool_ref.channel_registry.get_or_create(
            _power_managing._Report,  # pylint: disable=protected-access
            sub.get_channel_name(),
        )
        channel.resend_latest = True

        # More details on why the cast is needed here:
        # https://github.com/frequenz-floss/frequenz-sdk-python/issues/823
        return typing.cast(ReceiverFetcher[PVPoolReport], channel)

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

    @property
    def _system_power_bounds(self) -> ReceiverFetcher[SystemBounds]:
        """Return a receiver fetcher for the system power bounds."""
        return self._pv_pool_ref.bounds_channel
Attributes¤
component_ids property ¤
component_ids: Set[int]

Return component IDs of all PV inverters managed by this PVPool.

RETURNS DESCRIPTION
Set[int]

Set of managed component IDs.

power property ¤

Fetch the total power for the PV Inverters in the pool.

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

If a formula engine to calculate PV Inverter power is not already running, it will be started.

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

RETURNS DESCRIPTION
FormulaEngine[Power]

A FormulaEngine that will calculate and stream the total power of all PV Inverters.

power_status property ¤

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 batteries, - the result of the last distribution request for the pool's set of batteries.

RETURNS DESCRIPTION
ReceiverFetcher[PVPoolReport]

A receiver that will stream power status reports for the pool's priority.

Functions¤
__init__ ¤
__init__(
    pv_pool_ref: PVPoolReferenceStore,
    name: str | None,
    priority: int,
) -> None

Initialize the instance.

Note

PVPool instances are not meant to be created directly by users. Use the microgrid.pv_pool method for creating PVPool instances.

PARAMETER DESCRIPTION
pv_pool_ref

The reference store for the PV pool.

TYPE: PVPoolReferenceStore

name

The name of the PV pool.

TYPE: str | None

priority

The priority of the PV pool.

TYPE: int

Source code in frequenz/sdk/timeseries/pv_pool/_pv_pool.py
def __init__(  # pylint: disable=too-many-arguments
    self,
    pv_pool_ref: PVPoolReferenceStore,
    name: str | None,
    priority: int,
) -> None:
    """Initialize the instance.

    !!! note
        `PVPool` instances are not meant to be created directly by users. Use the
        [`microgrid.pv_pool`][frequenz.sdk.microgrid.pv_pool] method for creating
        `PVPool` instances.

    Args:
        pv_pool_ref: The reference store for the PV pool.
        name: The name of the PV pool.
        priority: The priority of the PV pool.
    """
    self._pv_pool_ref = pv_pool_ref
    unique_id = uuid.uuid4()
    self._source_id = str(unique_id) if name is None else f"{name}-{unique_id}"
    self._priority = priority
propose_power async ¤
propose_power(
    power: Power | None,
    *,
    request_timeout: timedelta = timedelta(seconds=5.0),
    bounds: Bounds[Power | None] = Bounds(None, None)
) -> None

Send a proposal to the power manager for the pool's set of PV inverters.

This proposal is for the maximum power that can be set for the PV inverters in the pool. The actual production might be lower.

Power values need to follow the Passive Sign Convention (PSC). That is, positive values indicate charge power and negative values indicate discharge power. Only discharge powers are allowed for PV inverters.

If the same PV inverters are shared by multiple actors, the power manager will consider the priority of the actors, the bounds they set, and their preferred power, when calculating the target power for the PV inverters.

The preferred power of lower priority actors will take precedence as long as they respect the bounds set by higher priority actors. If lower priority actors request power values outside of the bounds set by higher priority actors, the target power will be the closest value to the preferred power that is within the bounds.

When there are no other actors trying to use the same PV inverters, the actor's preferred power would be set as the target power, as long as it falls within the system power bounds for the PV inverters.

The result of the request can be accessed using the receiver returned from the power_status method, which also streams the bounds that an actor should comply with, based on its priority.

PARAMETER DESCRIPTION
power

The power to propose for the PV inverters in the pool. If None, this proposal will not have any effect on the target power, unless bounds are specified. If both are None, it is equivalent to not having a proposal or withdrawing a previous one.

TYPE: Power | None

request_timeout

The timeout for the request.

TYPE: timedelta DEFAULT: timedelta(seconds=5.0)

bounds

The power bounds for the proposal. These bounds will apply to actors with a lower priority, and can be overridden by bounds from actors with a higher priority. If None, the power bounds will be set to the maximum power of the batteries in the pool. This is currently and experimental feature.

TYPE: Bounds[Power | None] DEFAULT: Bounds(None, None)

RAISES DESCRIPTION
PVPoolError

If a charge power for PV inverters is requested.

Source code in frequenz/sdk/timeseries/pv_pool/_pv_pool.py
async def propose_power(
    self,
    power: Power | None,
    *,
    request_timeout: timedelta = timedelta(seconds=5.0),
    bounds: Bounds[Power | None] = Bounds(None, None),
) -> None:
    """Send a proposal to the power manager for the pool's set of PV inverters.

    This proposal is for the maximum power that can be set for the PV inverters in
    the pool.  The actual production might be lower.

    Power values need to follow the Passive Sign Convention (PSC). That is, positive
    values indicate charge power and negative values indicate discharge power.
    Only discharge powers are allowed for PV inverters.

    If the same PV inverters are shared by multiple actors, the power manager will
    consider the priority of the actors, the bounds they set, and their preferred
    power, when calculating the target power for the PV inverters.

    The preferred power of lower priority actors will take precedence as long as
    they respect the bounds set by higher priority actors.  If lower priority actors
    request power values outside of the bounds set by higher priority actors, the
    target power will be the closest value to the preferred power that is within the
    bounds.

    When there are no other actors trying to use the same PV inverters, the actor's
    preferred power would be set as the target power, as long as it falls within the
    system power bounds for the PV inverters.

    The result of the request can be accessed using the receiver returned from the
    [`power_status`][frequenz.sdk.timeseries.pv_pool.PVPool.power_status]
    method, which also streams the bounds that an actor should comply with, based on
    its priority.

    Args:
        power: The power to propose for the PV inverters in the pool.  If `None`,
            this proposal will not have any effect on the target power, unless
            bounds are specified.  If both are `None`, it is equivalent to not
            having a proposal or withdrawing a previous one.
        request_timeout: The timeout for the request.
        bounds: The power bounds for the proposal.  These bounds will apply to
            actors with a lower priority, and can be overridden by bounds from
            actors with a higher priority.  If None, the power bounds will be set to
            the maximum power of the batteries in the pool.  This is currently and
            experimental feature.

    Raises:
        PVPoolError: If a charge power for PV inverters is requested.
    """
    if power is not None and power > Power.zero():
        raise PVPoolError("Charge powers for PV inverters is not supported.")
    await self._pv_pool_ref.power_manager_requests_sender.send(
        _power_managing.Proposal(
            source_id=self._source_id,
            preferred_power=power,
            bounds=bounds,
            component_ids=self._pv_pool_ref.component_ids,
            priority=self._priority,
            creation_time=asyncio.get_running_loop().time(),
            request_timeout=request_timeout,
        )
    )
stop async ¤
stop() -> None

Stop all tasks and channels owned by the PVPool.

Source code in frequenz/sdk/timeseries/pv_pool/_pv_pool.py
async def stop(self) -> None:
    """Stop all tasks and channels owned by the PVPool."""
    await self._pv_pool_ref.stop()

frequenz.sdk.timeseries.pv_pool.PVPoolError ¤

Bases: Exception

An error that occurred in any of the PVPool methods.

Source code in frequenz/sdk/timeseries/pv_pool/_pv_pool.py
class PVPoolError(Exception):
    """An error that occurred in any of the PVPool methods."""

frequenz.sdk.timeseries.pv_pool.PVPoolReport ¤

Bases: Protocol

A status report for a PV pool.

Source code in frequenz/sdk/timeseries/pv_pool/_result_types.py
class PVPoolReport(typing.Protocol):
    """A status report for a PV pool."""

    target_power: Power | None
    """The currently set power for the PV inverters."""

    distribution_result: power_distributing.Result | None
    """The result of the last power distribution.

    This is `None` if no power distribution has been performed yet.
    """

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

        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 PV inverters.

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

distribution_result instance-attribute ¤
distribution_result: Result | None

The result of the last power distribution.

This is None if no power distribution has been performed yet.

target_power instance-attribute ¤
target_power: Power | None

The currently set power for the PV inverters.