Initialize grid in data pipeline

The grid is initialized in the data pipeline, so that it can
instantiate the formula engine pool to generate grid metrics.
This caused circular imports, which is resolved by enclosing
the type hints in a `TYPE_CHECKING` block.

Also the component metadata is now retrieved from the component
graph, instead of the microgrid API client. This is because
asynchronous calls to the API are not suitable in the data
pipeline as this is a high level user facing interface.

The grid initialize() function needed to be changed to process
the metadata as a dictionary rather than as a ComponentMetadata,
and this is due to the fact that the component graph does not
effectively retrieve the metadata as a ComponentMetadata object.

Signed-off-by: Daniel Zullo <[email protected]>

Author: daniel-zullo-frequenz

src/frequenz/sdk/microgrid/__init__.py

@@ -123,7 +123,6 @@
 """  # noqa: D205, D400
 
 from ..actor import ResamplerConfig
-from ..timeseries.grid import initialize as initialize_grid
 from . import _data_pipeline, client, component, connection_manager, metadata
 from ._data_pipeline import (
     battery_pool,
@@ -143,11 +142,6 @@ async def initialize(host: str, port: int, resampler_config: ResamplerConfig) ->
         resampler_config: Configuration for the resampling actor.
     """
     await connection_manager.initialize(host, port)
-
-    api_client = connection_manager.get().api_client
-    components = await api_client.components()
-    initialize_grid(components)
-
     await _data_pipeline.initialize(resampler_config)
 
 

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -24,6 +24,7 @@
 from ..timeseries._grid_frequency import GridFrequency
 from ..timeseries.grid import Grid
 from ..timeseries.grid import get as get_grid
+from ..timeseries.grid import initialize as initialize_grid
 from . import connection_manager
 from .component import ComponentCategory
 
@@ -115,6 +116,7 @@ def __init__(
         self._power_managing_actor: _power_managing.PowerManagingActor | None = None
 
         self._logical_meter: LogicalMeter | None = None
+        self._grid: Grid | None = None
         self._ev_charger_pools: dict[frozenset[int], EVChargerPool] = {}
         self._battery_pools: dict[frozenset[int], BatteryPoolReferenceStore] = {}
         self._frequency_pool: dict[int, GridFrequency] = {}
@@ -199,7 +201,14 @@ def grid(
         Returns:
             A Grid instance.
         """
-        return get_grid()
+        if self._grid is None:
+            initialize_grid(
+                channel_registry=self._channel_registry,
+                resampler_subscription_sender=self._resampling_request_sender(),
+            )
+            self._grid = get_grid()
+
+        return self._grid
 
     def battery_pool(
         self,

src/frequenz/sdk/timeseries/formula_engine/_formula_engine_pool.py

@@ -9,7 +9,6 @@
 
 from frequenz.channels import Sender
 
-from ...actor import ChannelRegistry, ComponentMetricRequest
 from ...microgrid.component import ComponentMetricId
 from .._quantities import Current, Power, Quantity
 from ._formula_generators._formula_generator import (
@@ -20,6 +19,7 @@
 
 if TYPE_CHECKING:
     # Break circular import
+    from ...actor import ChannelRegistry, ComponentMetricRequest
     from ..formula_engine import FormulaEngine, FormulaEngine3Phase
 
 

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

@@ -3,23 +3,27 @@
 
 """Base class for formula generators that use the component graphs."""
 
+from __future__ import annotations
 
 import sys
 from abc import ABC, abstractmethod
 from collections import abc
 from collections.abc import Callable
 from dataclasses import dataclass
-from typing import Generic
+from typing import TYPE_CHECKING, Generic
 
 from frequenz.channels import Sender
 
-from ....actor import ChannelRegistry, ComponentMetricRequest
 from ....microgrid import component, connection_manager
 from ....microgrid.component import ComponentMetricId
 from ..._quantities import QuantityT
 from .._formula_engine import FormulaEngine, FormulaEngine3Phase
 from .._resampled_formula_builder import ResampledFormulaBuilder
 
+if TYPE_CHECKING:
+    # Break circular import
+    from ....actor import ChannelRegistry, ComponentMetricRequest
+
 
 class FormulaGenerationError(Exception):
     """An error encountered during formula generation from the component graph."""

src/frequenz/sdk/timeseries/formula_engine/_resampled_formula_builder.py

@@ -3,19 +3,23 @@
 
 """A builder for creating formula engines that operate on resampled component metrics."""
 
+from __future__ import annotations
 
 from collections.abc import Callable
-from typing import Generic
+from typing import TYPE_CHECKING, Generic
 
 from frequenz.channels import Receiver, Sender
 
-from ...actor import ChannelRegistry, ComponentMetricRequest
 from ...microgrid.component import ComponentMetricId
 from .. import Sample
 from .._quantities import QuantityT
 from ._formula_engine import FormulaBuilder, FormulaEngine
 from ._tokenizer import Tokenizer, TokenType
 
+if TYPE_CHECKING:
+    # Break circular import
+    from ...actor import ChannelRegistry, ComponentMetricRequest
+
 
 class ResampledFormulaBuilder(Generic[QuantityT], FormulaBuilder[QuantityT]):
     """Provides a way to build a FormulaEngine from resampled data streams."""
@@ -63,6 +67,9 @@ def _get_resampled_receiver(
         Returns:
             A receiver to stream resampled data for the given component id.
         """
+        # pylint: disable=import-outside-toplevel
+        from frequenz.sdk.actor import ComponentMetricRequest
+
         request = ComponentMetricRequest(self._namespace, component_id, metric_id, None)
         self._resampler_requests.append(request)
         return self._channel_registry.new_receiver(request.get_channel_name())

src/frequenz/sdk/timeseries/grid.py

@@ -6,16 +6,16 @@
 This module provides the `Grid` type, which represents a grid connection point
 in a microgrid.
 """
+from __future__ import annotations
 
 import logging
 import uuid
-from collections.abc import Iterable
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
 from frequenz.channels import Sender
 
-from ..microgrid.component import Component
+from ..microgrid import connection_manager
 from ..microgrid.component._component import ComponentCategory
 from . import Fuse
 from .formula_engine._formula_engine_pool import FormulaEnginePool
@@ -42,14 +42,12 @@ class Grid:
 
 
 def initialize(
-    components: Iterable[Component],
     channel_registry: ChannelRegistry,
     resampler_subscription_sender: Sender[ComponentMetricRequest],
 ) -> None:
     """Initialize the grid connection.
 
     Args:
-        components: The components in the microgrid.
         channel_registry: The channel registry instance shared with the
             resampling actor.
         resampler_subscription_sender: The sender for sending metric requests
@@ -63,9 +61,9 @@ def initialize(
     global _GRID  # pylint: disable=global-statement
 
     grid_connections = list(
-        component
-        for component in components
-        if component.category == ComponentCategory.GRID
+        connection_manager.get().component_graph.components(
+            component_category={ComponentCategory.GRID},
+        )
     )
 
     grid_connections_count = len(grid_connections)
@@ -82,7 +80,17 @@ def initialize(
         if grid_connections[0].metadata is None:
             raise RuntimeError("Grid metadata is None")
 
-        fuse = grid_connections[0].metadata.fuse
+        fuse: Fuse | None = None
+
+        # The current implementation of the Component Graph fails to
+        # effectively convert components from a dictionary representation to
+        # the expected Component object.
+        # Specifically for the component metadata, it hands back a dictionary
+        # instead of the expected ComponentMetadata type.
+        metadata = grid_connections[0].metadata
+        if isinstance(metadata, dict):
+            fuse_dict = metadata.get("fuse", None)
+            fuse = Fuse(**fuse_dict) if fuse_dict else None
 
         if fuse is None:
             raise RuntimeError("Grid fuse is None")