Skip to content

component_graph

frequenz.sdk.microgrid.component_graph ¤

Defines a graph representation of how microgrid components are connected.

The component graph is an approximate representation of the microgrid circuit, abstracted to a level appropriate for higher-level monitoring and control. Examples of use-cases would be:

  • using the graph structure to infer which component measurements need to be combined to obtain grid power or onsite load

  • identifying which inverter(s) need to be engaged to (dis)charge a particular battery

  • understanding which power flows in the microgrid are derived from green and grey sources

It deliberately does not include all pieces of hardware placed in the microgrid, instead limiting itself to just those that are needed to monitor and control the flow of power.

Classes¤

frequenz.sdk.microgrid.component_graph.ComponentGraph ¤

Bases: ABC

Interface for component graph implementations.

Source code in frequenz/sdk/microgrid/component_graph.py
class ComponentGraph(ABC):
    """Interface for component graph implementations."""

    @abstractmethod
    def components(
        self,
        component_ids: set[int] | None = None,
        component_categories: set[ComponentCategory] | None = None,
    ) -> set[Component]:
        """Fetch the components of the microgrid.

        Args:
            component_ids: filter out any components not matching one of the provided IDs
            component_categories: filter out any components not matching one of the
                provided types

        Returns:
            Set of the components currently connected to the microgrid, filtered by
                the provided `component_ids` and `component_categories` values.
        """

    @abstractmethod
    def connections(
        self,
        start: set[int] | None = None,
        end: set[int] | None = None,
    ) -> set[Connection]:
        """Fetch the connections between microgrid components.

        Args:
            start: filter out any connections whose `start` does not match one of these
                component IDs
            end: filter out any connections whose `end` does not match one of these
                component IDs

        Returns:
            Set of the connections between components in the microgrid, filtered by
                the provided `start`/`end` choices.
        """

    @abstractmethod
    def predecessors(self, component_id: int) -> set[Component]:
        """Fetch the graph predecessors of the specified component.

        Args:
            component_id: numerical ID of the component whose predecessors should be
                fetched

        Returns:
            Set of IDs of the components that are predecessors of `component_id`,
                i.e. for which there is a connection from each of these components to
                `component_id`.

        Raises:
            KeyError: if the specified `component_id` is not in the graph
        """

    @abstractmethod
    def successors(self, component_id: int) -> set[Component]:
        """Fetch the graph successors of the specified component.

        Args:
            component_id: numerical ID of the component whose successors should be
                fetched

        Returns:
            Set of IDs of the components that are successors of `component_id`,
                i.e. for which there is a connection from `component_id` to each of
                these components.

        Raises:
            KeyError: if the specified `component_id` is not in the graph
        """

    @abstractmethod
    def is_grid_meter(self, component: Component) -> bool:
        """Check if the specified component is a grid meter.

        This is done by checking if the component is the only successor to the `Grid`
        component.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is a grid meter.
        """

    @abstractmethod
    def is_pv_inverter(self, component: Component) -> bool:
        """Check if the specified component is a PV inverter.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is a PV inverter.
        """

    @abstractmethod
    def is_pv_meter(self, component: Component) -> bool:
        """Check if the specified component is a PV meter.

        This is done by checking if the component has only PV inverters as its
        successors.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is a PV meter.
        """

    @abstractmethod
    def is_pv_chain(self, component: Component) -> bool:
        """Check if the specified component is part of a PV chain.

        A component is part of a PV chain if it is a PV meter or a PV inverter.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is part of a PV chain.
        """

    @abstractmethod
    def is_battery_inverter(self, component: Component) -> bool:
        """Check if the specified component is a battery inverter.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is a battery inverter.
        """

    @abstractmethod
    def is_battery_meter(self, component: Component) -> bool:
        """Check if the specified component is a battery meter.

        This is done by checking if the component has only battery inverters as its
        predecessors.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is a battery meter.
        """

    @abstractmethod
    def is_battery_chain(self, component: Component) -> bool:
        """Check if the specified component is part of a battery chain.

        A component is part of a battery chain if it is a battery meter or a battery
        inverter.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is part of a battery chain.
        """

    @abstractmethod
    def is_ev_charger(self, component: Component) -> bool:
        """Check if the specified component is an EV charger.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is an EV charger.
        """

    @abstractmethod
    def is_ev_charger_meter(self, component: Component) -> bool:
        """Check if the specified component is an EV charger meter.

        This is done by checking if the component has only EV chargers as its
        successors.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is an EV charger meter.
        """

    @abstractmethod
    def is_ev_charger_chain(self, component: Component) -> bool:
        """Check if the specified component is part of an EV charger chain.

        A component is part of an EV charger chain if it is an EV charger meter or an
        EV charger.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is part of an EV charger chain.
        """

    @abstractmethod
    def is_chp(self, component: Component) -> bool:
        """Check if the specified component is a CHP.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is a CHP.
        """

    @abstractmethod
    def is_chp_meter(self, component: Component) -> bool:
        """Check if the specified component is a CHP meter.

        This is done by checking if the component has only CHPs as its successors.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is a CHP meter.
        """

    @abstractmethod
    def is_chp_chain(self, component: Component) -> bool:
        """Check if the specified component is part of a CHP chain.

        A component is part of a CHP chain if it is a CHP meter or a CHP.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is part of a CHP chain.
        """

    @abstractmethod
    def dfs(
        self,
        current_node: Component,
        visited: set[Component],
        condition: Callable[[Component], bool],
    ) -> set[Component]:
        """
        Search for components that fulfill the condition in the Graph.

        DFS is used for searching the graph. The graph traversal is stopped
        once a component fulfills the condition.

        Args:
            current_node: The current node to search from.
            visited: The set of visited nodes.
            condition: The condition function to check for.

        Returns:
            A set of component ids where the corresponding components fulfill
            the condition function.
        """

    @abstractmethod
    def find_first_descendant_component(
        self,
        *,
        root_category: ComponentCategory,
        descendant_categories: Iterable[ComponentCategory],
    ) -> Component:
        """Find the first descendant component given root and descendant categories.

        This method searches for the root component within the provided root
        category. If multiple components share the same root category, the
        first found one is considered as the root component.

        Subsequently, it looks for the first descendant component from the root
        component, considering only the immediate descendants.

        The priority of the component to search for is determined by the order
        of the descendant categories, with the first category having the
        highest priority.

        Args:
            root_category: The category of the root component to search for.
            descendant_categories: The descendant categories to search for the
                first descendant component in.

        Returns:
            The first descendant component found in the component graph,
            considering the specified root and descendant categories.
        """
Functions¤
components abstractmethod ¤
components(
    component_ids: set[int] | None = None,
    component_categories: (
        set[ComponentCategory] | None
    ) = None,
) -> set[Component]

Fetch the components of the microgrid.

PARAMETER DESCRIPTION
component_ids

filter out any components not matching one of the provided IDs

TYPE: set[int] | None DEFAULT: None

component_categories

filter out any components not matching one of the provided types

TYPE: set[ComponentCategory] | None DEFAULT: None

RETURNS DESCRIPTION
set[Component]

Set of the components currently connected to the microgrid, filtered by the provided component_ids and component_categories values.

Source code in frequenz/sdk/microgrid/component_graph.py
@abstractmethod
def components(
    self,
    component_ids: set[int] | None = None,
    component_categories: set[ComponentCategory] | None = None,
) -> set[Component]:
    """Fetch the components of the microgrid.

    Args:
        component_ids: filter out any components not matching one of the provided IDs
        component_categories: filter out any components not matching one of the
            provided types

    Returns:
        Set of the components currently connected to the microgrid, filtered by
            the provided `component_ids` and `component_categories` values.
    """
connections abstractmethod ¤
connections(
    start: set[int] | None = None,
    end: set[int] | None = None,
) -> set[Connection]

Fetch the connections between microgrid components.

PARAMETER DESCRIPTION
start

filter out any connections whose start does not match one of these component IDs

TYPE: set[int] | None DEFAULT: None

end

filter out any connections whose end does not match one of these component IDs

TYPE: set[int] | None DEFAULT: None

RETURNS DESCRIPTION
set[Connection]

Set of the connections between components in the microgrid, filtered by the provided start/end choices.

Source code in frequenz/sdk/microgrid/component_graph.py
@abstractmethod
def connections(
    self,
    start: set[int] | None = None,
    end: set[int] | None = None,
) -> set[Connection]:
    """Fetch the connections between microgrid components.

    Args:
        start: filter out any connections whose `start` does not match one of these
            component IDs
        end: filter out any connections whose `end` does not match one of these
            component IDs

    Returns:
        Set of the connections between components in the microgrid, filtered by
            the provided `start`/`end` choices.
    """
dfs abstractmethod ¤
dfs(
    current_node: Component,
    visited: set[Component],
    condition: Callable[[Component], bool],
) -> set[Component]

Search for components that fulfill the condition in the Graph.

DFS is used for searching the graph. The graph traversal is stopped once a component fulfills the condition.

PARAMETER DESCRIPTION
current_node

The current node to search from.

TYPE: Component

visited

The set of visited nodes.

TYPE: set[Component]

condition

The condition function to check for.

TYPE: Callable[[Component], bool]

RETURNS DESCRIPTION
set[Component]

A set of component ids where the corresponding components fulfill

set[Component]

the condition function.

Source code in frequenz/sdk/microgrid/component_graph.py
@abstractmethod
def dfs(
    self,
    current_node: Component,
    visited: set[Component],
    condition: Callable[[Component], bool],
) -> set[Component]:
    """
    Search for components that fulfill the condition in the Graph.

    DFS is used for searching the graph. The graph traversal is stopped
    once a component fulfills the condition.

    Args:
        current_node: The current node to search from.
        visited: The set of visited nodes.
        condition: The condition function to check for.

    Returns:
        A set of component ids where the corresponding components fulfill
        the condition function.
    """
find_first_descendant_component abstractmethod ¤
find_first_descendant_component(
    *,
    root_category: ComponentCategory,
    descendant_categories: Iterable[ComponentCategory]
) -> Component

Find the first descendant component given root and descendant categories.

This method searches for the root component within the provided root category. If multiple components share the same root category, the first found one is considered as the root component.

Subsequently, it looks for the first descendant component from the root component, considering only the immediate descendants.

The priority of the component to search for is determined by the order of the descendant categories, with the first category having the highest priority.

PARAMETER DESCRIPTION
root_category

The category of the root component to search for.

TYPE: ComponentCategory

descendant_categories

The descendant categories to search for the first descendant component in.

TYPE: Iterable[ComponentCategory]

RETURNS DESCRIPTION
Component

The first descendant component found in the component graph,

Component

considering the specified root and descendant categories.

Source code in frequenz/sdk/microgrid/component_graph.py
@abstractmethod
def find_first_descendant_component(
    self,
    *,
    root_category: ComponentCategory,
    descendant_categories: Iterable[ComponentCategory],
) -> Component:
    """Find the first descendant component given root and descendant categories.

    This method searches for the root component within the provided root
    category. If multiple components share the same root category, the
    first found one is considered as the root component.

    Subsequently, it looks for the first descendant component from the root
    component, considering only the immediate descendants.

    The priority of the component to search for is determined by the order
    of the descendant categories, with the first category having the
    highest priority.

    Args:
        root_category: The category of the root component to search for.
        descendant_categories: The descendant categories to search for the
            first descendant component in.

    Returns:
        The first descendant component found in the component graph,
        considering the specified root and descendant categories.
    """
is_battery_chain abstractmethod ¤
is_battery_chain(component: Component) -> bool

Check if the specified component is part of a battery chain.

A component is part of a battery chain if it is a battery meter or a battery inverter.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is part of a battery chain.

Source code in frequenz/sdk/microgrid/component_graph.py
@abstractmethod
def is_battery_chain(self, component: Component) -> bool:
    """Check if the specified component is part of a battery chain.

    A component is part of a battery chain if it is a battery meter or a battery
    inverter.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is part of a battery chain.
    """
is_battery_inverter abstractmethod ¤
is_battery_inverter(component: Component) -> bool

Check if the specified component is a battery inverter.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is a battery inverter.

Source code in frequenz/sdk/microgrid/component_graph.py
@abstractmethod
def is_battery_inverter(self, component: Component) -> bool:
    """Check if the specified component is a battery inverter.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is a battery inverter.
    """
is_battery_meter abstractmethod ¤
is_battery_meter(component: Component) -> bool

Check if the specified component is a battery meter.

This is done by checking if the component has only battery inverters as its predecessors.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is a battery meter.

Source code in frequenz/sdk/microgrid/component_graph.py
@abstractmethod
def is_battery_meter(self, component: Component) -> bool:
    """Check if the specified component is a battery meter.

    This is done by checking if the component has only battery inverters as its
    predecessors.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is a battery meter.
    """
is_chp abstractmethod ¤
is_chp(component: Component) -> bool

Check if the specified component is a CHP.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is a CHP.

Source code in frequenz/sdk/microgrid/component_graph.py
@abstractmethod
def is_chp(self, component: Component) -> bool:
    """Check if the specified component is a CHP.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is a CHP.
    """
is_chp_chain abstractmethod ¤
is_chp_chain(component: Component) -> bool

Check if the specified component is part of a CHP chain.

A component is part of a CHP chain if it is a CHP meter or a CHP.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is part of a CHP chain.

Source code in frequenz/sdk/microgrid/component_graph.py
@abstractmethod
def is_chp_chain(self, component: Component) -> bool:
    """Check if the specified component is part of a CHP chain.

    A component is part of a CHP chain if it is a CHP meter or a CHP.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is part of a CHP chain.
    """
is_chp_meter abstractmethod ¤
is_chp_meter(component: Component) -> bool

Check if the specified component is a CHP meter.

This is done by checking if the component has only CHPs as its successors.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is a CHP meter.

Source code in frequenz/sdk/microgrid/component_graph.py
@abstractmethod
def is_chp_meter(self, component: Component) -> bool:
    """Check if the specified component is a CHP meter.

    This is done by checking if the component has only CHPs as its successors.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is a CHP meter.
    """
is_ev_charger abstractmethod ¤
is_ev_charger(component: Component) -> bool

Check if the specified component is an EV charger.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is an EV charger.

Source code in frequenz/sdk/microgrid/component_graph.py
@abstractmethod
def is_ev_charger(self, component: Component) -> bool:
    """Check if the specified component is an EV charger.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is an EV charger.
    """
is_ev_charger_chain abstractmethod ¤
is_ev_charger_chain(component: Component) -> bool

Check if the specified component is part of an EV charger chain.

A component is part of an EV charger chain if it is an EV charger meter or an EV charger.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is part of an EV charger chain.

Source code in frequenz/sdk/microgrid/component_graph.py
@abstractmethod
def is_ev_charger_chain(self, component: Component) -> bool:
    """Check if the specified component is part of an EV charger chain.

    A component is part of an EV charger chain if it is an EV charger meter or an
    EV charger.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is part of an EV charger chain.
    """
is_ev_charger_meter abstractmethod ¤
is_ev_charger_meter(component: Component) -> bool

Check if the specified component is an EV charger meter.

This is done by checking if the component has only EV chargers as its successors.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is an EV charger meter.

Source code in frequenz/sdk/microgrid/component_graph.py
@abstractmethod
def is_ev_charger_meter(self, component: Component) -> bool:
    """Check if the specified component is an EV charger meter.

    This is done by checking if the component has only EV chargers as its
    successors.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is an EV charger meter.
    """
is_grid_meter abstractmethod ¤
is_grid_meter(component: Component) -> bool

Check if the specified component is a grid meter.

This is done by checking if the component is the only successor to the Grid component.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is a grid meter.

Source code in frequenz/sdk/microgrid/component_graph.py
@abstractmethod
def is_grid_meter(self, component: Component) -> bool:
    """Check if the specified component is a grid meter.

    This is done by checking if the component is the only successor to the `Grid`
    component.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is a grid meter.
    """
is_pv_chain abstractmethod ¤
is_pv_chain(component: Component) -> bool

Check if the specified component is part of a PV chain.

A component is part of a PV chain if it is a PV meter or a PV inverter.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is part of a PV chain.

Source code in frequenz/sdk/microgrid/component_graph.py
@abstractmethod
def is_pv_chain(self, component: Component) -> bool:
    """Check if the specified component is part of a PV chain.

    A component is part of a PV chain if it is a PV meter or a PV inverter.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is part of a PV chain.
    """
is_pv_inverter abstractmethod ¤
is_pv_inverter(component: Component) -> bool

Check if the specified component is a PV inverter.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is a PV inverter.

Source code in frequenz/sdk/microgrid/component_graph.py
@abstractmethod
def is_pv_inverter(self, component: Component) -> bool:
    """Check if the specified component is a PV inverter.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is a PV inverter.
    """
is_pv_meter abstractmethod ¤
is_pv_meter(component: Component) -> bool

Check if the specified component is a PV meter.

This is done by checking if the component has only PV inverters as its successors.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is a PV meter.

Source code in frequenz/sdk/microgrid/component_graph.py
@abstractmethod
def is_pv_meter(self, component: Component) -> bool:
    """Check if the specified component is a PV meter.

    This is done by checking if the component has only PV inverters as its
    successors.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is a PV meter.
    """
predecessors abstractmethod ¤
predecessors(component_id: int) -> set[Component]

Fetch the graph predecessors of the specified component.

PARAMETER DESCRIPTION
component_id

numerical ID of the component whose predecessors should be fetched

TYPE: int

RETURNS DESCRIPTION
set[Component]

Set of IDs of the components that are predecessors of component_id, i.e. for which there is a connection from each of these components to component_id.

RAISES DESCRIPTION
KeyError

if the specified component_id is not in the graph

Source code in frequenz/sdk/microgrid/component_graph.py
@abstractmethod
def predecessors(self, component_id: int) -> set[Component]:
    """Fetch the graph predecessors of the specified component.

    Args:
        component_id: numerical ID of the component whose predecessors should be
            fetched

    Returns:
        Set of IDs of the components that are predecessors of `component_id`,
            i.e. for which there is a connection from each of these components to
            `component_id`.

    Raises:
        KeyError: if the specified `component_id` is not in the graph
    """
successors abstractmethod ¤
successors(component_id: int) -> set[Component]

Fetch the graph successors of the specified component.

PARAMETER DESCRIPTION
component_id

numerical ID of the component whose successors should be fetched

TYPE: int

RETURNS DESCRIPTION
set[Component]

Set of IDs of the components that are successors of component_id, i.e. for which there is a connection from component_id to each of these components.

RAISES DESCRIPTION
KeyError

if the specified component_id is not in the graph

Source code in frequenz/sdk/microgrid/component_graph.py
@abstractmethod
def successors(self, component_id: int) -> set[Component]:
    """Fetch the graph successors of the specified component.

    Args:
        component_id: numerical ID of the component whose successors should be
            fetched

    Returns:
        Set of IDs of the components that are successors of `component_id`,
            i.e. for which there is a connection from `component_id` to each of
            these components.

    Raises:
        KeyError: if the specified `component_id` is not in the graph
    """

frequenz.sdk.microgrid.component_graph.InvalidGraphError ¤

Bases: Exception

Exception type that will be thrown if graph data is not valid.

Source code in frequenz/sdk/microgrid/component_graph.py
class InvalidGraphError(Exception):
    """Exception type that will be thrown if graph data is not valid."""