Add ComponentGraph wrapper adding back the component type checks

That's how it used to be before the rust component graph was
introduced.

Signed-off-by: Sahas Subramanian <[email protected]>

Author: shsms

pyproject.toml

@@ -29,8 +29,8 @@ dependencies = [
   # Make sure to update the mkdocs.yml file when
   # changing the version
   # (plugins.mkdocstrings.handlers.python.import)
-  "frequenz-client-microgrid >= 0.18.0, < 0.19.0",
-  "frequenz-microgrid-component-graph >= 0.2.0, < 0.3",
+  "frequenz-client-microgrid >= 0.18.1, < 0.19.0",
+  "frequenz-microgrid-component-graph >= 0.3.2, < 0.4",
   "frequenz-client-common >= 0.3.6, < 0.4.0",
   "frequenz-channels >= 1.6.1, < 2.0.0",
   "frequenz-quantities[marshmallow] >= 1.0.0, < 2.0.0",

src/frequenz/sdk/microgrid/component_graph.py

@@ -0,0 +1,175 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Component graph representation for a microgrid."""
+
+from frequenz.client.common.microgrid.components import ComponentId
+from frequenz.client.microgrid.component import (
+    BatteryInverter,
+    Chp,
+    Component,
+    ComponentConnection,
+    EvCharger,
+    SolarInverter,
+)
+from frequenz.microgrid_component_graph import ComponentGraph as BaseComponentGraph
+from typing_extensions import override
+
+
+class ComponentGraph(BaseComponentGraph[Component, ComponentConnection, ComponentId]):
+    """A representation of a microgrid's component graph."""
+
+    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.
+        """
+        return isinstance(component, SolarInverter)
+
+    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 either a PV inverter or a PV
+        meter.
+
+        Args:
+            component: The component to check.
+
+        Returns:
+            Whether the specified component is part of a PV chain.
+        """
+        return self.is_pv_inverter(component) or self.is_pv_meter(component)
+
+    @override
+    def is_pv_meter(self, component: Component | ComponentId) -> bool:
+        """Check if the specified component is a PV meter.
+
+        Args:
+            component: The component or component ID to check.
+
+        Returns:
+            Whether the specified component is a PV meter.
+        """
+        if isinstance(component, Component):
+            return super().is_pv_meter(component.id)
+        return super().is_pv_meter(component)
+
+    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.
+        """
+        return isinstance(component, EvCharger)
+
+    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 either an EV charger or an
+        EV charger meter.
+
+        Args:
+            component: The component to check.
+
+        Returns:
+            Whether the specified component is part of an EV charger chain.
+        """
+        return self.is_ev_charger(component) or self.is_ev_charger_meter(component)
+
+    @override
+    def is_ev_charger_meter(self, component: Component | ComponentId) -> bool:
+        """Check if the specified component is an EV charger meter.
+
+        Args:
+            component: The component or component ID to check.
+
+        Returns:
+            Whether the specified component is an EV charger meter.
+        """
+        if isinstance(component, Component):
+            return super().is_ev_charger_meter(component.id)
+        return super().is_ev_charger_meter(component)
+
+    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.
+        """
+        return isinstance(component, BatteryInverter)
+
+    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 either a battery inverter or a
+        battery meter.
+
+        Args:
+            component: The component to check.
+
+        Returns:
+            Whether the specified component is part of a battery chain.
+        """
+        return self.is_battery_inverter(component) or self.is_battery_meter(component)
+
+    @override
+    def is_battery_meter(self, component: Component | ComponentId) -> bool:
+        """Check if the specified component is a battery meter.
+
+        Args:
+            component: The component or component ID to check.
+
+        Returns:
+            Whether the specified component is a battery meter.
+        """
+        if isinstance(component, Component):
+            return super().is_battery_meter(component.id)
+        return super().is_battery_meter(component)
+
+    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.
+        """
+        return isinstance(component, Chp)
+
+    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 either a CHP or a CHP meter.
+
+        Args:
+            component: The component to check.
+
+        Returns:
+            Whether the specified component is part of a CHP chain.
+        """
+        return self.is_chp(component) or self.is_chp_meter(component)
+
+    @override
+    def is_chp_meter(self, component: Component | ComponentId) -> bool:
+        """Check if the specified component is a CHP meter.
+
+        Args:
+            component: The component or component ID to check.
+
+        Returns:
+            Whether the specified component is a CHP meter.
+        """
+        if isinstance(component, Component):
+            return super().is_chp_meter(component.id)
+        return super().is_chp_meter(component)

src/frequenz/sdk/microgrid/connection_manager.py

@@ -13,14 +13,13 @@
 from abc import ABC, abstractmethod
 
 from frequenz.client.common.microgrid import MicrogridId
-from frequenz.client.common.microgrid.components import ComponentId
 from frequenz.client.microgrid import (
     Location,
     MicrogridApiClient,
     MicrogridInfo,
 )
-from frequenz.client.microgrid.component import Component, ComponentConnection
-from frequenz.microgrid_component_graph import ComponentGraph
+
+from .component_graph import ComponentGraph
 
 _logger = logging.getLogger(__name__)
 
@@ -59,7 +58,7 @@ def api_client(self) -> MicrogridApiClient:
     @abstractmethod
     def component_graph(
         self,
-    ) -> ComponentGraph[Component, ComponentConnection, ComponentId]:
+    ) -> ComponentGraph:
         """Get component graph.
 
         Returns:
@@ -109,9 +108,7 @@ def __init__(self, server_url: str) -> None:
         self._client = MicrogridApiClient(server_url)
         # To create graph from the API client we need await.
         # So create empty graph here, and update it in `run` method.
-        self._graph: (
-            ComponentGraph[Component, ComponentConnection, ComponentId] | None
-        ) = None
+        self._graph: ComponentGraph | None = None
 
         self._microgrid: MicrogridInfo
         """The microgrid information."""
@@ -140,9 +137,7 @@ def location(self) -> Location | None:
         return self._microgrid.location
 
     @property
-    def component_graph(
-        self,
-    ) -> ComponentGraph[Component, ComponentConnection, ComponentId]:
+    def component_graph(self) -> ComponentGraph:
         """Get component graph.
 
         Returns: