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. Common use cases include:

  • Combining component measurements to compute grid power or onsite load by using the graph structure to determine which measurements to aggregate

  • Identifying which inverter(s) need to be engaged to charge or discharge a particular battery based on their connectivity in the graph

  • Understanding which power flows in the microgrid are derived from green vs grey sources based on the component connectivity

The graph 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: The component IDs that the components must match.
            component_categories: The component categories that the components must match.

        Returns:
            The set of 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: The component IDs that the connections' start must match.
            end: The component IDs that the connections' end must match.

        Returns:
            The set of 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: The IDs of the components whose predecessors should be
                fetched.

        Returns:
            The set of 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: The IDs of the components whose successors should be fetched.

        Returns:
            The set of 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: The 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: The 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: The 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: The 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: The 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: The 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: The 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: The 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: The 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: The 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: The 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: The 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: The 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 class of the root component to search for.
            descendant_categories: The descendant classes to search for the first
                descendant component in.

        Returns:
            The first descendant component found in the component graph,
                considering the specified `root` and `descendants` 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

The component IDs that the components must match.

TYPE: set[int] | None DEFAULT: None

component_categories

The component categories that the components must match.

TYPE: set[ComponentCategory] | None DEFAULT: None

RETURNS DESCRIPTION
set[Component]

The set of 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: The component IDs that the components must match.
        component_categories: The component categories that the components must match.

    Returns:
        The set of 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

The component IDs that the connections' start must match.

TYPE: set[int] | None DEFAULT: None

end

The component IDs that the connections' end must match.

TYPE: set[int] | None DEFAULT: None

RETURNS DESCRIPTION
set[Connection]

The set of 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: The component IDs that the connections' start must match.
        end: The component IDs that the connections' end must match.

    Returns:
        The set of 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 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 class of the root component to search for.

TYPE: ComponentCategory

descendant_categories

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

TYPE: Iterable[ComponentCategory]

RETURNS DESCRIPTION
Component

The first descendant component found in the component graph, considering the specified root and descendants 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 class of the root component to search for.
        descendant_categories: The descendant classes to search for the first
            descendant component in.

    Returns:
        The first descendant component found in the component graph,
            considering the specified `root` and `descendants` 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

The 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: The 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

The 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: The 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

The 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: The 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

The 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: The 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

The 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: The 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

The 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: The 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

The 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: The 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

The 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: The 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

The 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: The 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

The 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: The 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

The 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: The 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

The 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: The 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

The 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: The 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

The IDs of the components whose predecessors should be fetched.

TYPE: int

RETURNS DESCRIPTION
set[Component]

The set of 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: The IDs of the components whose predecessors should be
            fetched.

    Returns:
        The set of 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

The IDs of the components whose successors should be fetched.

TYPE: int

RETURNS DESCRIPTION
set[Component]

The set of 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: The IDs of the components whose successors should be fetched.

    Returns:
        The set of 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."""