Add depth first search for components (#595)

In this PR we add a dfs component search to the component graph. The
search continues to go into depth until the condition is fulfilled. The
condition has to be provided as a lambda.

Furthermore we add function that returns the grid component such that it
can be used in the dfs search, whenever the search should begin at the
`GRID` component.

Author: matthias-wende-frequenz

RELEASE_NOTES.md

@@ -10,7 +10,7 @@
 
 ## New Features
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+- DFS for compentent graph
 
 ## Bug Fixes
 

src/frequenz/sdk/microgrid/_graph.py

@@ -266,6 +266,29 @@ def is_chp_chain(self, component: Component) -> bool:
             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 travarsal 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 coresponding components fulfill
+            the condition function.
+        """
+
 
 class _MicrogridComponentGraph(ComponentGraph):
     """ComponentGraph implementation designed to work with the microgrid API.
@@ -688,6 +711,42 @@ def is_chp_chain(self, component: Component) -> bool:
         """
         return self.is_chp(component) or self.is_chp_meter(component)
 
+    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 travarsal 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 coresponding components fulfill
+            the condition function.
+        """
+        if current_node in visited:
+            return set()
+
+        visited.add(current_node)
+
+        if condition(current_node):
+            return {current_node}
+
+        component: Set[Component] = set()
+
+        for successor in self.successors(current_node.component_id):
+            component.update(self.dfs(successor, visited, condition))
+
+        return component
+
     def _validate_graph(self) -> None:
         """Check that the underlying graph data is valid.
 

src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_formula_generator.py

@@ -101,15 +101,15 @@ def _get_builder(
         )
         return builder
 
-    def _get_grid_component_successors(self) -> set[component.Component]:
-        """Get the set of grid component successors in the component graph.
+    def _get_grid_component(self) -> component.Component:
+        """
+        Get the grid component in the component graph.
 
         Returns:
-            A set of grid component successors.
+            The first grid component found in the graph.
 
         Raises:
             ComponentNotFound: If the grid component is not found in the component graph.
-            ComponentNotFound: If no successor components are found in the component graph.
         """
         component_graph = connection_manager.get().component_graph
         grid_component = next(
@@ -120,10 +120,22 @@ def _get_grid_component_successors(self) -> set[component.Component]:
             ),
             None,
         )
-
         if grid_component is None:
             raise ComponentNotFound("Grid component not found in the component graph.")
 
+        return grid_component
+
+    def _get_grid_component_successors(self) -> set[component.Component]:
+        """Get the set of grid component successors in the component graph.
+
+        Returns:
+            A set of grid component successors.
+
+        Raises:
+            ComponentNotFound: If no successor components are found in the component graph.
+        """
+        grid_component = self._get_grid_component()
+        component_graph = connection_manager.get().component_graph
         grid_successors = component_graph.successors(grid_component.component_id)
 
         if not grid_successors:

tests/microgrid/test_graph.py

@@ -403,6 +403,116 @@ def test_connection_filters(self) -> None:
             Connection(2, 6),
         }
 
+    def test_dfs_search_two_grid_meters(self) -> None:
+        """Test DFS searching PV components in a graph with two grid meters."""
+        grid = Component(1, ComponentCategory.GRID)
+        pv_inverters = {
+            Component(4, ComponentCategory.INVERTER, InverterType.SOLAR),
+            Component(5, ComponentCategory.INVERTER, InverterType.SOLAR),
+        }
+
+        graph = gr._MicrogridComponentGraph(
+            components={
+                grid,
+                Component(2, ComponentCategory.METER),
+                Component(3, ComponentCategory.METER),
+            }.union(pv_inverters),
+            connections={
+                Connection(1, 2),
+                Connection(1, 3),
+                Connection(2, 4),
+                Connection(2, 5),
+            },
+        )
+
+        result = graph.dfs(grid, set(), graph.is_pv_inverter)
+        assert result == pv_inverters
+
+    def test_dfs_search_grid_meter(self) -> None:
+        """Test DFS searching PV components in a graph with a single grid meter."""
+        grid = Component(1, ComponentCategory.GRID)
+        pv_meters = {
+            Component(3, ComponentCategory.METER),
+            Component(4, ComponentCategory.METER),
+        }
+
+        graph = gr._MicrogridComponentGraph(
+            components={
+                grid,
+                Component(2, ComponentCategory.METER),
+                Component(5, ComponentCategory.INVERTER, InverterType.SOLAR),
+                Component(6, ComponentCategory.INVERTER, InverterType.SOLAR),
+            }.union(pv_meters),
+            connections={
+                Connection(1, 2),
+                Connection(2, 3),
+                Connection(2, 4),
+                Connection(3, 5),
+                Connection(4, 6),
+            },
+        )
+
+        result = graph.dfs(grid, set(), graph.is_pv_chain)
+        assert result == pv_meters
+
+    def test_dfs_search_no_grid_meter(self) -> None:
+        """Test DFS searching PV components in a graph with no grid meter."""
+        grid = Component(1, ComponentCategory.GRID)
+        pv_meters = {
+            Component(3, ComponentCategory.METER),
+            Component(4, ComponentCategory.METER),
+        }
+
+        graph = gr._MicrogridComponentGraph(
+            components={
+                grid,
+                Component(2, ComponentCategory.METER),
+                Component(5, ComponentCategory.INVERTER, InverterType.SOLAR),
+                Component(6, ComponentCategory.INVERTER, InverterType.SOLAR),
+            }.union(pv_meters),
+            connections={
+                Connection(1, 2),
+                Connection(1, 3),
+                Connection(1, 4),
+                Connection(3, 5),
+                Connection(4, 6),
+            },
+        )
+
+        result = graph.dfs(grid, set(), graph.is_pv_chain)
+        assert result == pv_meters
+
+    def test_dfs_search_nested_components(self) -> None:
+        """Test DFS searching PV components in a graph with nested components."""
+        grid = Component(1, ComponentCategory.GRID)
+        battery_components = {
+            Component(4, ComponentCategory.METER),
+            Component(5, ComponentCategory.METER),
+            Component(6, ComponentCategory.INVERTER, InverterType.BATTERY),
+        }
+
+        graph = gr._MicrogridComponentGraph(
+            components={
+                grid,
+                Component(2, ComponentCategory.METER),
+                Component(3, ComponentCategory.METER),
+                Component(7, ComponentCategory.INVERTER, InverterType.BATTERY),
+                Component(8, ComponentCategory.INVERTER, InverterType.BATTERY),
+            }.union(battery_components),
+            connections={
+                Connection(1, 2),
+                Connection(2, 3),
+                Connection(2, 6),
+                Connection(3, 4),
+                Connection(3, 5),
+                Connection(4, 7),
+                Connection(5, 8),
+            },
+        )
+
+        assert set() == graph.dfs(grid, set(), graph.is_pv_chain)
+        assert battery_components == graph.dfs(grid, set(), graph.is_battery_chain)
+
 
 class Test_MicrogridComponentGraph:
     """Test cases for the package-internal implementation of the ComponentGraph.