Replace uses of ComponentCategory with the new types

The new API client provides a mapping of `ComponentCategory` and
`XxxType`s to Python classes. This commit updates most uses of component
categories with these classes.

This also includes some updates of imports for `Component` and
`ComponentCategory` and some changes on specific component type, like
`GridConnectionPoint`'s metadata.

Finally, the component graph interface to get components was also
updated to use types instead of categories, for which the API was
slightly changed to better represent the filtering criteria. The
keyword for arguments were changed from `component_ids` to
`matching_ids` and `component_categories` to `matching_types`.

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

benchmarks/power_distribution/power_distributor.py

@@ -13,7 +13,7 @@
 
 from frequenz.channels import Broadcast
 from frequenz.client.common.microgrid.components import ComponentId
-from frequenz.client.microgrid import Component, ComponentCategory
+from frequenz.client.microgrid.component import Battery
 from frequenz.quantities import Power
 
 from frequenz.sdk import microgrid
@@ -116,8 +116,7 @@ async def run_test(  # pylint: disable=too-many-locals
     battery_status_channel = Broadcast[ComponentPoolStatus](name="battery-status")
     power_result_channel = Broadcast[Result](name="power-result")
     async with PowerDistributingActor(
-        component_category=ComponentCategory.BATTERY,
-        component_type=None,
+        component_type=Battery,
         requests_receiver=power_request_channel.new_receiver(),
         results_sender=power_result_channel.new_sender(),
         component_pool_status_sender=battery_status_channel.new_sender(),
@@ -143,8 +142,8 @@ async def run() -> None:
         ResamplerConfig2(resampling_period=timedelta(seconds=1.0)),
     )
 
-    all_batteries: set[Component] = connection_manager.get().component_graph.components(
-        component_categories={ComponentCategory.BATTERY}
+    all_batteries = connection_manager.get().component_graph.components(
+        matching_types={Battery},
     )
     batteries_ids = {c.id for c in all_batteries}
     # Take some time to get data from components

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -18,7 +18,7 @@
 
 from frequenz.channels import Broadcast, Sender
 from frequenz.client.common.microgrid.components import ComponentId
-from frequenz.client.microgrid import ComponentCategory, InverterType
+from frequenz.client.microgrid.component import Battery, EvCharger, SolarInverter
 
 from frequenz.sdk.microgrid._power_managing._base_classes import Algorithm, DefaultPower
 
@@ -105,23 +105,25 @@ def __init__(
             self._channel_registry,
             api_power_request_timeout=api_power_request_timeout,
             power_manager_algorithm=Algorithm.SHIFTING_MATRYOSHKA,
-            component_category=ComponentCategory.BATTERY,
             default_power=DefaultPower.ZERO,
+            component_class=Battery,
         )
         self._ev_power_wrapper = PowerWrapper(
             self._channel_registry,
             api_power_request_timeout=api_power_request_timeout,
             power_manager_algorithm=Algorithm.MATRYOSHKA,
-            component_category=ComponentCategory.EV_CHARGER,
             default_power=DefaultPower.MAX,
+            component_class=EvCharger,
         )
         self._pv_power_wrapper = PowerWrapper(
             self._channel_registry,
             api_power_request_timeout=api_power_request_timeout,
             power_manager_algorithm=Algorithm.MATRYOSHKA,
-            component_category=ComponentCategory.INVERTER,
-            component_type=InverterType.SOLAR,
             default_power=DefaultPower.MIN,
+            # Using SolarInverter might be too specific, maybe we need to also pass
+            # HybridInverter, see
+            # https://github.com/frequenz-floss/frequenz-sdk-python/issues/1285
+            component_class=SolarInverter,
         )
 
         self._logical_meter: LogicalMeter | None = None

src/frequenz/sdk/microgrid/_old_component_data.py

@@ -160,7 +160,7 @@ def _check_category(cls, component_id: ComponentId) -> None:
         from .. import microgrid
 
         components = microgrid.connection_manager.get().component_graph.components(
-            filter_by_ids={component_id}
+            matching_ids={component_id}
         )
         if not components:
             raise ValueError(f"Unable to find component with {component_id}")

src/frequenz/sdk/microgrid/_power_distributing/_component_managers/_battery_manager.py

@@ -12,11 +12,8 @@
 
 from frequenz.channels import LatestValueCache, Receiver, Sender
 from frequenz.client.common.microgrid.components import ComponentId
-from frequenz.client.microgrid import (
-    ApiClientError,
-    ComponentCategory,
-    OperationOutOfRange,
-)
+from frequenz.client.microgrid import ApiClientError, OperationOutOfRange
+from frequenz.client.microgrid.component import Battery, Inverter
 from frequenz.quantities import Power
 from typing_extensions import override
 
@@ -85,7 +82,9 @@ def _get_battery_inverter_mappings(
         inverters: set[ComponentId] = set(
             component.id
             for component in component_graph.predecessors(battery_id)
-            if component.category == ComponentCategory.INVERTER
+            # Using Inverter is way too general, see
+            # https://github.com/frequenz-floss/frequenz-sdk-python/issues/1285
+            if isinstance(component, Inverter)
         )
 
         if len(inverters) == 0:
@@ -147,7 +146,7 @@ def __init__(
         self._results_sender = results_sender
         self._api_power_request_timeout = api_power_request_timeout
         self._batteries = connection_manager.get().component_graph.components(
-            component_categories={ComponentCategory.BATTERY}
+            matching_types={Battery}
         )
         self._battery_ids = {battery.id for battery in self._batteries}
 

src/frequenz/sdk/microgrid/_power_distributing/_component_managers/_ev_charger_manager/_ev_charger_manager.py

@@ -17,11 +17,8 @@
     selected_from,
 )
 from frequenz.client.common.microgrid.components import ComponentId
-from frequenz.client.microgrid import (
-    ApiClientError,
-    ComponentCategory,
-    MicrogridApiClient,
-)
+from frequenz.client.microgrid import ApiClientError, MicrogridApiClient
+from frequenz.client.microgrid.component import EvCharger
 from frequenz.quantities import Power, Voltage
 from typing_extensions import override
 
@@ -115,7 +112,7 @@ def _get_ev_charger_ids(self) -> collections.abc.Set[ComponentId]:
         return {
             evc.id
             for evc in connection_manager.get().component_graph.components(
-                component_categories={ComponentCategory.EV_CHARGER}
+                matching_types={EvCharger}
             )
         }
 

src/frequenz/sdk/microgrid/_power_distributing/_component_managers/_pv_inverter_manager/_pv_inverter_manager.py

@@ -10,11 +10,8 @@
 
 from frequenz.channels import LatestValueCache, Sender
 from frequenz.client.common.microgrid.components import ComponentId
-from frequenz.client.microgrid import (
-    ApiClientError,
-    ComponentCategory,
-    InverterType,
-)
+from frequenz.client.microgrid import ApiClientError
+from frequenz.client.microgrid.component import SolarInverter
 from frequenz.quantities import Power
 from typing_extensions import override
 
@@ -258,7 +255,6 @@ def _get_pv_inverter_ids(self) -> collections.abc.Set[ComponentId]:
         return {
             inv.id
             for inv in connection_manager.get().component_graph.components(
-                component_categories={ComponentCategory.INVERTER}
+                matching_types={SolarInverter}
             )
-            if inv.type == InverterType.SOLAR
         }

src/frequenz/sdk/microgrid/_power_distributing/_component_status/_battery_status_tracker.py

@@ -27,10 +27,10 @@
 from frequenz.client.microgrid import (
     BatteryComponentState,
     BatteryRelayState,
-    ComponentCategory,
     ErrorLevel,
     InverterComponentState,
 )
+from frequenz.client.microgrid.component import Inverter
 from typing_extensions import override
 
 from frequenz.sdk._internal._asyncio import run_forever
@@ -486,7 +486,9 @@ def _find_adjacent_inverter_id(self, battery_id: ComponentId) -> ComponentId | N
             (
                 comp.id
                 for comp in graph.predecessors(battery_id)
-                if comp.category == ComponentCategory.INVERTER
+                # Using Inverter is way too general, see
+                # https://github.com/frequenz-floss/frequenz-sdk-python/issues/1285
+                if isinstance(comp, Inverter)
             ),
             None,
         )

src/frequenz/sdk/microgrid/_power_distributing/power_distributing.py

@@ -13,10 +13,11 @@
 import asyncio
 import logging
 from datetime import datetime, timedelta
+from typing import assert_never
 
 from frequenz.channels import Receiver, Sender
 from frequenz.client.common.microgrid.components import ComponentId
-from frequenz.client.microgrid import ComponentCategory, ComponentType, InverterType
+from frequenz.client.microgrid.component import Battery, EvCharger, SolarInverter
 from typing_extensions import override
 
 from ...actor._actor import Actor
@@ -60,43 +61,31 @@ class PowerDistributingActor(Actor):  # pylint: disable=too-many-instance-attrib
 
     def __init__(  # pylint: disable=too-many-arguments
         self,
+        component_type: type[Battery | EvCharger | SolarInverter],
         requests_receiver: Receiver[Request],
         results_sender: Sender[Result],
         component_pool_status_sender: Sender[ComponentPoolStatus],
         *,
         api_power_request_timeout: timedelta,
-        component_category: ComponentCategory,
-        component_type: ComponentType | None = None,
         name: str | None = None,
     ) -> None:
         """Create actor instance.
 
         Args:
+            component_type: The class of the components that this actor is
+                responsible for.
             requests_receiver: Receiver for receiving power requests from the power
                 manager.
             results_sender: Sender for sending results to the power manager.
             component_pool_status_sender: Channel for sending information about which
                 components are expected to be working.
             api_power_request_timeout: Timeout to use when making power requests to
                 the microgrid API.
-            component_category: The category of the components that this actor is
-                responsible for.
-            component_type: The type of the component of the given category that this
-                actor is responsible for.  This is used only when the component category
-                is not enough to uniquely identify the component.  For example, when the
-                category is `ComponentCategory.INVERTER`, the type is needed to identify
-                the inverter as a solar inverter or a battery inverter.  This can be
-                `None` when the component category is enough to uniquely identify the
-                component.
             name: The name of the actor. If `None`, `str(id(self))` will be used. This
                 is used mostly for debugging purposes.
-
-        Raises:
-            ValueError: If the given component category is not supported.
         """
         super().__init__(name=name)
-        self._component_category = component_category
-        self._component_type = component_type
+        self._component_class = component_type
         self._requests_receiver = requests_receiver
         self._result_sender = results_sender
         self._api_power_request_timeout = api_power_request_timeout
@@ -114,25 +103,20 @@ def __init__(  # pylint: disable=too-many-arguments
         """
 
         self._component_manager: ComponentManager
-        if component_category == ComponentCategory.BATTERY:
+        if issubclass(component_type, Battery):
             self._component_manager = BatteryManager(
                 component_pool_status_sender, results_sender, api_power_request_timeout
             )
-        elif component_category == ComponentCategory.EV_CHARGER:
+        elif issubclass(component_type, EvCharger):
             self._component_manager = EVChargerManager(
                 component_pool_status_sender, results_sender, api_power_request_timeout
             )
-        elif (
-            component_category == ComponentCategory.INVERTER
-            and component_type == InverterType.SOLAR
-        ):
+        elif issubclass(component_type, SolarInverter):
             self._component_manager = PVManager(
                 component_pool_status_sender, results_sender, api_power_request_timeout
             )
         else:
-            raise ValueError(
-                f"PowerDistributor doesn't support controlling: {component_category}"
-            )
+            assert_never(component_type)
 
     @override
     async def _run(self) -> None:

src/frequenz/sdk/microgrid/_power_managing/_power_managing_actor.py

@@ -14,7 +14,7 @@
 from frequenz.channels import Receiver, Sender, select, selected_from
 from frequenz.channels.timer import SkipMissedAndDrift, Timer
 from frequenz.client.common.microgrid.components import ComponentId
-from frequenz.client.microgrid import ComponentCategory, ComponentType, InverterType
+from frequenz.client.microgrid.component import Battery, EvCharger, SolarInverter
 from typing_extensions import override
 
 from ..._internal._asyncio import run_forever
@@ -49,8 +49,7 @@ def __init__(  # pylint: disable=too-many-arguments
         channel_registry: ChannelRegistry,
         algorithm: Algorithm,
         default_power: DefaultPower,
-        component_category: ComponentCategory,
-        component_type: ComponentType | None = None,
+        component_class: type[Battery | EvCharger | SolarInverter],
     ):
         """Create a new instance of the power manager.
 
@@ -64,19 +63,10 @@ def __init__(  # pylint: disable=too-many-arguments
             channel_registry: The channel registry.
             algorithm: The power management algorithm to use.
             default_power: The default power to use for the components.
-            component_category: The category of the component this power manager
-                instance is going to support.
-            component_type: The type of the component of the given category that this
-                actor is responsible for.  This is used only when the component category
-                is not enough to uniquely identify the component.  For example, when the
-                category is `ComponentCategory.INVERTER`, the type is needed to identify
-                the inverter as a solar inverter or a battery inverter.  This can be
-                `None` when the component category is enough to uniquely identify the
-                component.
+            component_class: The class of component this instance is going to support.
         """
-        self._component_category = component_category
-        self._component_type = component_type
         self._default_power = default_power
+        self._component_class = component_class
         self._bounds_subscription_receiver = bounds_subscription_receiver
         self._power_distributing_requests_sender = power_distributing_requests_sender
         self._power_distributing_results_receiver = power_distributing_results_receiver
@@ -153,39 +143,32 @@ def _add_system_bounds_tracker(self, component_ids: frozenset[ComponentId]) -> N
 
         Args:
             component_ids: The component IDs for which to add a bounds tracker.
-
-        Raises:
-            NotImplementedError: When the pool type is not supported.
         """
         bounds_receiver: Receiver[SystemBounds]
-        if self._component_category is ComponentCategory.BATTERY:
+        if issubclass(self._component_class, Battery):
             battery_pool = _data_pipeline.new_battery_pool(
                 priority=-sys.maxsize - 1, component_ids=component_ids
             )
             # pylint: disable-next=protected-access
             bounds_receiver = battery_pool._system_power_bounds.new_receiver()
-        elif self._component_category is ComponentCategory.EV_CHARGER:
+        elif issubclass(self._component_class, EvCharger):
             ev_charger_pool = _data_pipeline.new_ev_charger_pool(
                 priority=-sys.maxsize - 1, component_ids=component_ids
             )
             # pylint: disable-next=protected-access
             bounds_receiver = ev_charger_pool._system_power_bounds.new_receiver()
-        elif (
-            self._component_category is ComponentCategory.INVERTER
-            and self._component_type is InverterType.SOLAR
-        ):
+        elif issubclass(self._component_class, SolarInverter):
             pv_pool = _data_pipeline.new_pv_pool(
                 priority=-sys.maxsize - 1, component_ids=component_ids
             )
             # pylint: disable-next=protected-access
             bounds_receiver = pv_pool._system_power_bounds.new_receiver()
         else:
-            err = (
-                "PowerManagingActor: Unsupported component category: "
-                f"{self._component_category}"
+            _logger.error(
+                "PowerManagingActor: Unsupported component class: %s",
+                self._component_class.__name__,
             )
-            _logger.error(err)
-            raise NotImplementedError(err)
+            assert_never(self._component_class)
 
         self._system_bounds[component_ids] = SystemBounds(
             timestamp=datetime.now(tz=timezone.utc),