Skip to content

ev_charger_pool

frequenz.sdk.timeseries.ev_charger_pool ¤

Interactions with EV Chargers.

Classes¤

frequenz.sdk.timeseries.ev_charger_pool.EVChargerPool ¤

Bases: ComponentPool[EVChargerPoolReferenceStore, EVChargerPoolReport]

An interface for interaction with pools of EV Chargers.

Provides
Source code in src/frequenz/sdk/timeseries/ev_charger_pool/_ev_charger_pool.py 
 21
 22
 23
 24
 25
 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
class EVChargerPool(ComponentPool[EVChargerPoolReferenceStore, EVChargerPoolReport]):
    """An interface for interaction with pools of EV Chargers.

    Provides:
      - Aggregate [`power`][frequenz.sdk.timeseries.ev_charger_pool.EVChargerPool.power]
        and
        [`current_per_phase`][frequenz.sdk.timeseries.ev_charger_pool.EVChargerPool.current_per_phase]
        measurements of the EV Chargers 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 EV chargers.

        This proposal is for the maximum power that can be set for the EV chargers in
        the pool.  The actual consumption might be lower based on the number of phases
        an EV is drawing power from, and its current state of charge.

        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 EV chargers 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:
            EVChargerPoolError: If a discharge power for EV chargers is requested.
        """
        if power is not None and power < Power.zero():
            raise EVChargerPoolError(
                "Discharging from EV chargers is currently not supported."
            )
        await super().propose_power(power, bounds=bounds)

    @property
    def current_per_phase(self) -> Formula3Phase[Current]:
        """Fetch the total current for the EV Chargers in the pool.

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

        If a formula to calculate EV Charger current 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 current of all EV
                Chargers.
        """
        return self._pool_ref_store.formula_pool.from_current_3_phase_formula(
            "ev_charger_total_current",
            connection_manager.get().component_graph.ev_charger_formula(
                self._pool_ref_store.component_ids
            ),
        )

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

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

        If a formula to calculate EV Charger 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 EV
                Chargers.
        """
        return self._pool_ref_store.formula_pool.from_power_formula(
            "ev_charger_power",
            connection_manager.get().component_graph.ev_charger_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.
current_per_phase property ¤
current_per_phase: Formula3Phase[Current]

Fetch the total current for the EV Chargers in the pool.

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

If a formula to calculate EV Charger current is not already running, it will be started.

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

RETURNS DESCRIPTION
Formula3Phase[Current]

A Formula that will calculate and stream the total current of all EV Chargers.
power property ¤
power: Formula[Power]

Fetch the total power for the EV Chargers in the pool.

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

If a formula to calculate EV Charger 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 EV Chargers.
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.

Functions¤
__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 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
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 EV chargers.

This proposal is for the maximum power that can be set for the EV chargers in the pool. The actual consumption might be lower based on the number of phases an EV is drawing power from, and its current state of charge.

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

PARAMETER DESCRIPTION
power

The power to propose for the EV chargers 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
EVChargerPoolError

If a discharge power for EV chargers is requested.
Source code in src/frequenz/sdk/timeseries/ev_charger_pool/_ev_charger_pool.py 
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
@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 EV chargers.

    This proposal is for the maximum power that can be set for the EV chargers in
    the pool.  The actual consumption might be lower based on the number of phases
    an EV is drawing power from, and its current state of charge.

    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 EV chargers 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:
        EVChargerPoolError: If a discharge power for EV chargers is requested.
    """
    if power is not None and power < Power.zero():
        raise EVChargerPoolError(
            "Discharging from EV chargers is currently 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 
153
154
async def stop(self) -> None:
    """Stop all tasks and channels owned by the pool."""

frequenz.sdk.timeseries.ev_charger_pool.EVChargerPoolError ¤

Bases: Exception

An error that occurred in any of the EVChargerPool methods.

Source code in src/frequenz/sdk/timeseries/ev_charger_pool/_ev_charger_pool.py 
17
18
class EVChargerPoolError(Exception):
    """An error that occurred in any of the EVChargerPool methods."""

frequenz.sdk.timeseries.ev_charger_pool.EVChargerPoolReport ¤

Bases: ComponentPoolReport, Protocol

A status report for an EV chargers pool.

Source code in src/frequenz/sdk/timeseries/ev_charger_pool/_result_types.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class EVChargerPoolReport(ComponentPoolReport, typing.Protocol):
    """A status report for an EV chargers pool."""

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

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

        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 EV chargers.

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 EV chargers.