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 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
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,
        *,
        pool_ref_store: 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.new_pv_pool`][frequenz.sdk.microgrid.new_pv_pool] method for
            creating `PVPool` instances.

        Args:
            pool_ref_store: The reference store for the PV pool.
            name: The name of the PV pool.
            priority: The priority of the PV pool.
        """
        self._pool_ref_store = pool_ref_store
        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,
        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.

        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 PV inverters in the pool.  If `None`,
                this proposal will not have any effect on the target power, unless
                bounds are specified.  When speficied 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, this will limit
                the bounds for lower priority actors.

        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._pool_ref_store.power_manager_requests_sender.send(
            _power_managing.Proposal(
                source_id=self._source_id,
                preferred_power=power,
                bounds=bounds,
                component_ids=self._pool_ref_store.component_ids,
                priority=self._priority,
                creation_time=asyncio.get_running_loop().time(),
            )
        )

    @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._pool_ref_store.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._pool_ref_store.formula_pool.from_power_formula_generator(
            "pv_power",
            PVPowerFormula,
            FormulaGeneratorConfig(
                component_ids=self._pool_ref_store.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._pool_ref_store.component_ids,
        )
        self._pool_ref_store.power_bounds_subs[sub.get_channel_name()] = (
            asyncio.create_task(
                self._pool_ref_store.power_manager_bounds_subs_sender.send(sub)
            )
        )
        channel = self._pool_ref_store.channel_registry.get_or_create(
            _power_managing._Report,  # pylint: disable=protected-access
            sub.get_channel_name(),
        )
        channel.resend_latest = True

        return channel

    @property
    def power_distribution_results(self) -> ReceiverFetcher[_power_distributing.Result]:
        """Get a receiver to receive power distribution results.

        Returns:
            A receiver that will stream power distribution results for the pool's set of
            PV inverters.
        """
        return MappingReceiverFetcher(
            self._pool_ref_store.power_distribution_results_fetcher,
            lambda recv: recv.filter(
                lambda x: x.request.component_ids == self._pool_ref_store.component_ids
            ),
        )

    async def stop(self) -> None:
        """Stop all tasks and channels owned by the PVPool."""
        # This was closing the pool_ref_store, which is not correct, because those are
        # shared.
        #
        # This method will do until we have a mechanism to track the resources created
        # through it.  It can also eventually cleanup the pool_ref_store, when it is
        # holding the last reference to it.

    @property
    def _system_power_bounds(self) -> ReceiverFetcher[SystemBounds]:
        """Return a receiver fetcher for the system power bounds."""
        return self._pool_ref_store.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 ¤
power: 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 DESCRIPTION
FormulaEngine[Power]

A FormulaEngine that will calculate and stream the total power of all PV Inverters.
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]

PV inverters.
power_status property ¤
power_status: 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 DESCRIPTION
ReceiverFetcher[PVPoolReport]

A receiver that will stream power status reports for the pool's priority.
Functions¤
__init__ ¤
__init__(
    *,
    pool_ref_store: 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.new_pv_pool method for creating PVPool instances.

PARAMETER DESCRIPTION
pool_ref_store

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 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def __init__(  # pylint: disable=too-many-arguments
    self,
    *,
    pool_ref_store: 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.new_pv_pool`][frequenz.sdk.microgrid.new_pv_pool] method for
        creating `PVPool` instances.

    Args:
        pool_ref_store: The reference store for the PV pool.
        name: The name of the PV pool.
        priority: The priority of the PV pool.
    """
    self._pool_ref_store = pool_ref_store
    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,
    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.

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

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. When speficied 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, this will limit the bounds for lower priority actors.

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 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
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 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.

    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 PV inverters in the pool.  If `None`,
            this proposal will not have any effect on the target power, unless
            bounds are specified.  When speficied 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, this will limit
            the bounds for lower priority actors.

    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._pool_ref_store.power_manager_requests_sender.send(
        _power_managing.Proposal(
            source_id=self._source_id,
            preferred_power=power,
            bounds=bounds,
            component_ids=self._pool_ref_store.component_ids,
            priority=self._priority,
            creation_time=asyncio.get_running_loop().time(),
        )
    )
stop async ¤
stop() -> None

Stop all tasks and channels owned by the PVPool.

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

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 
22
23
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 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
class PVPoolReport(typing.Protocol):
    """A status report for a PV pool."""

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

    @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.

target_power property ¤
target_power: Power | None

The currently set power for the PV inverters.