Initial client

Signed-off-by: Flora <[email protected]>

Author: flora-hofmann-frequenz

RELEASE_NOTES.md

@@ -2,15 +2,20 @@
 
 ## Summary
 
-<!-- Here goes a general summary of what this release is about -->
+This release introduces the initial version of the Assets API client, enabling retrieval of asset-related data.
 
 ## Upgrading
 
 <!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
 
 ## New Features
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+Introducing the initial version of the Assets API client, designed to streamline access to asset-related data.
+
+* Supports querying asset components and retrieving their metrics efficiently.
+* Provides a structured data representation while retaining raw protobuf responses.
+* Currently focused on retrieving asset data for individual components.
+* Examples are provided to guide users through the basic usage of the client.
 
 ## Bug Fixes
 

examples/client.py

@@ -0,0 +1,46 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Examples usage of PlatformAssets API."""
+
+import asyncio
+import argparse
+from pprint import pprint
+
+from frequenz.client.assets import AssetsApiClient
+
+
+async def main(microgrid_id: int) -> None:
+    """Test the AssetsApiClient.
+
+    Args:
+        microgrid_id: The ID of the microgrid to query.
+    """
+    server_url = "localhost:50052"
+    client = AssetsApiClient(server_url)
+
+    print("########################################################")
+    print("Fetching microgrid details")
+
+    microgrid_details = await client.get_microgrid(microgrid_id)
+    pprint(microgrid_details)
+
+    print("########################################################")
+    print("Listing microgrid components")
+
+    components = await client.list_microgrid_components(microgrid_id)
+    pprint(components)
+
+    print("########################################################")
+    print("Listing microgrid component connections")
+
+    connections = await client.list_microgrid_component_connections(microgrid_id)
+    pprint(connections)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("microgrid_id", type=int, help="Microgrid ID")
+
+    args = parser.parse_args()
+    asyncio.run(main(args.microgrid_id))

pyproject.toml

@@ -27,6 +27,8 @@ classifiers = [
 requires-python = ">= 3.11, < 4"
 dependencies = [
   "typing-extensions >= 4.12.2, < 5",
+  "frequenz-api-assets >= 0.1.0, < 1",
+  "frequenz-client-common >= 0.9.0, < 1",
 ]
 dynamic = ["version"]
 
@@ -82,6 +84,10 @@ dev = [
   "frequenz-client-assets[dev-mkdocs,dev-flake8,dev-formatting,dev-mkdocs,dev-mypy,dev-noxfile,dev-pylint,dev-pytest]",
 ]
 
+examples = [
+    "grpcio >= 1.51.1, < 2",
+]
+
 [project.urls]
 Documentation = "https://frequenz-floss.github.io/frequenz-client-assets-python/"
 Changelog = "https://github.com/frequenz-floss/frequenz-client-assets-python/releases"

src/frequenz/client/assets/__init__.py

@@ -1,22 +1,13 @@
 # License: MIT
 # Copyright © 2025 Frequenz Energy-as-a-Service GmbH
 
-"""Assets API client for Python."""
+"""Client to connect to the PlatformAssets API.
 
+This package provides a low-level interface for interacting
+with the assets API.
+"""
 
-# TODO(cookiecutter): Remove this function
-def delete_me(*, blow_up: bool = False) -> bool:
-    """Do stuff for demonstration purposes.
 
-    Args:
-        blow_up: If True, raise an exception.
+from ._client import AssetsApiClient
 
-    Returns:
-        True if no exception was raised.
-
-    Raises:
-        RuntimeError: if blow_up is True.
-    """
-    if blow_up:
-        raise RuntimeError("This function should be removed!")
-    return True
+__all__ = ["AssetsApiClient"]

src/frequenz/client/assets/_client.py

@@ -0,0 +1,94 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Client for requests to the PlatformAssets API."""
+
+from frequenz.client.base.client import BaseApiClient, call_stub_method
+from frequenz.client.base.exception import ClientNotConnected
+
+# pylint: disable=no-name-in-module
+from frequenz.api.assets.v1.assets_pb2_grpc import PlatformAssetsStub
+from frequenz.api.assets.v1.assets_pb2 import (
+    GetMicrogridRequest as PBGetMicrogridRequest)
+from frequenz.api.assets.v1.assets_pb2 import (
+    GetMicrogridResponse as PBGetMicrogridResponse)
+from frequenz.api.assets.v1.assets_pb2 import (
+    ListMicrogridComponentsRequest as PBListMicrogridComponentsRequest)
+from frequenz.api.assets.v1.assets_pb2 import (
+    ListMicrogridComponentsResponse as PBListMicrogridComponentsResponse)
+from frequenz.api.assets.v1.assets_pb2 import (
+    ListMicrogridComponentConnectionsRequest as PBListMicrogridComponentConnectionsRequest)
+from frequenz.api.assets.v1.assets_pb2 import (
+    ListMicrogridComponentConnectionsResponse as PBListMicrogridComponentConnectionsResponse)
+# pylint: enable=no-name-in-module
+
+class AssetsApiClient(BaseApiClient[PlatformAssetsStub]):
+    """A client for the Frequenz PlatformAssets service."""
+
+    def __init__(self, server_url: str, *, connect: bool = True) -> None:
+        """Initialize the client.
+
+        Args:
+            server_url: The URL of the PlatformAssets service.
+            connect: Whether to establish a connection immediately.
+        """
+        super().__init__(server_url, PlatformAssetsStub, connect=connect)
+
+    @property
+    def stub(self) -> PlatformAssetsStub:
+        """Return the gRPC stub for the PlatformAssets service."""
+        if self.channel is None or self._stub is None:
+            raise ClientNotConnected(server_url=self.server_url, operation="stub")
+        return self._stub  # type: ignore
+
+    async def get_microgrid(self, microgrid_id: int) -> PBGetMicrogridResponse:
+        """Fetch details of a specific microgrid.
+
+        Args:
+            microgrid_id: The ID of the microgrid to fetch.
+
+        Returns:
+            The response containing the microgrid details.
+        """
+        request = PBGetMicrogridRequest(microgrid_id=microgrid_id)
+        return await call_stub_method(self, lambda: self.stub.GetMicrogrid(request))
+
+    async def list_microgrid_components(
+        self, microgrid_id: int, component_ids=None, categories=None
+    ) -> PBListMicrogridComponentsResponse:
+        """List electrical components of a microgrid.
+
+        Args:
+            microgrid_id: The ID of the microgrid to fetch components for.
+            component_ids: The IDs of the components to fetch.
+            categories: The categories of the components to fetch.
+
+        Returns:
+            The response containing the microgrid components.
+        """
+        request = PBListMicrogridComponentsRequest(
+            microgrid_id=microgrid_id,
+            component_ids=component_ids or [],
+            categories=categories or [],
+        )
+        return await call_stub_method(self, lambda: self.stub.ListMicrogridComponents(request))
+
+    async def list_microgrid_component_connections(
+        self, microgrid_id: int, source_component_ids=None, destination_component_ids=None
+    ) -> PBListMicrogridComponentConnectionsResponse:
+        """List electrical connections between components in a microgrid.
+
+        Args:
+            microgrid_id: The ID of the microgrid to fetch component connections for.
+            source_component_ids: The IDs of the source components to fetch connections for.
+            destination_component_ids: The IDs of the destination components to fetch connections for.
+
+        Returns:
+            The response containing the component connections.
+        """
+        request = PBListMicrogridComponentConnectionsRequest(
+            microgrid_id=microgrid_id,
+            source_component_ids=source_component_ids or [],
+            destination_component_ids=destination_component_ids or [],
+        )
+        return await call_stub_method(self, lambda: self.stub.ListMicrogridComponentConnections(request))

tests/test_frequenz_client_assets.py

@@ -2,17 +2,32 @@
 # Copyright © 2025 Frequenz Energy-as-a-Service GmbH
 
 """Tests for the frequenz.client.assets package."""
-import pytest
+from typing import Generator
+from unittest.mock import MagicMock, patch
 
-from frequenz.client.assets import delete_me
+import pytest
 
+from frequenz.client.assets import AssetsApiClient
 
-def test_frequenz_assets_succeeds() -> None:  # TODO(cookiecutter): Remove
-    """Test that the delete_me function succeeds."""
-    assert delete_me() is True
+@pytest.fixture
+def mock_channel() -> Generator[MagicMock, None, None]:
+    """Fixture for grpc.aio.insecure_channel."""
+    with patch("grpc.aio.insecure_channel") as mock:
+        yield mock
 
+@pytest.mark.asyncio
+async def test_client_initialization(mock_channel: MagicMock) -> None:
+    """Test that the client initializes the channel."""
+    client = AssetsApiClient("localhost:50051")  # noqa: F841
+    mock_channel.assert_called_once_with("localhost:50051")
 
-def test_frequenz_assets_fails() -> None:  # TODO(cookiecutter): Remove
-    """Test that the delete_me function fails."""
-    with pytest.raises(RuntimeError, match="This function should be removed!"):
-        delete_me(blow_up=True)
+@pytest.mark.asyncio
+async def test_get_microgrid(mock_channel: MagicMock) -> None:
+    """Test that get_microgrid calls the stub correctly."""
+    mock_stub = MagicMock()
+    mock_stub.GetMicrogrid.return_value = MagicMock()
+    with patch("frequenz.client.assets.AssetsApiClient.stub", new_callable=lambda: mock_stub):
+        client = AssetsApiClient("localhost:50051")
+        response = await client.get_microgrid(123)
+        mock_stub.GetMicrogrid.assert_called_once()
+        assert response is not None