Enhance Assets API client with electrical components support and CLI improvements

This commit introduces several key updates to the Assets API client, including:
- New method `list_microgrid_electrical_components()` for retrieving electrical components in a microgrid.
- Updated CLI command `assets-cli electrical-components <microgrid-id>` for displaying electrical components.
- Improved error handling and type safety in the client.
- Enhanced JSON encoding for better serialization of enum keys.
- Added new exceptions for better error management.

These changes significantly enhance the functionality and usability of the Assets API client within the Frequenz microgrid framework.

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

Author: eduardiazf

RELEASE_NOTES.md

@@ -2,43 +2,24 @@
 
 ## Summary
 
-This release introduces a complete Assets API client with CLI support for interacting with Frequenz microgrid assets, including comprehensive error handling and type safety.
-
-## Upgrading
-
-**Breaking Changes:**
-
-- Added new required dependencies: `frequenz-api-assets`, `frequenz-api-common`, `frequenz-client-base`, `grpcio`
-
-**CLI Support:**
-Install with `pip install "frequenz-client-assets[cli]"` for command-line functionality.
+This release introduces a Assets API client with CLI support for interacting with Frequenz microgrid assets. It provides comprehensive electrical components functionality including batteries, EV chargers, inverters, and grid connection points, with enhanced type safety and error handling.
 
 ## New Features
 
-**Assets API Client:**
-
-- Complete gRPC client for Frequenz Assets API
-- Extends `BaseApiClient` for authentication and connection management
-- `get_microgrid_details()` method for retrieving microgrid information
+* **Assets API Client**:
+  * `list_electrical_components()` method for retrieving electrical components in a microgrid
 
-**Command-Line Interface:**
+* **Electrical Components Support**: Comprehensive data classes for electrical components
+  * `ElectricalComponent` with category-specific information for batteries, EV chargers, inverters, grid connection points, and power transformers
+  * Battery types: Li-ion, Na-ion with proper enum mapping
+  * EV charger types: AC, DC, Hybrid charging support
+  * Operational lifetime tracking and metric configuration bounds
 
-- `python -m frequenz.client.assets microgrid <id>` command
-- Environment variable support for API credentials
-- JSON output formatting
+* **Command-Line Interface**:
+  * `assets-cli electrical-components <microgrid-id>` command
 
-**Type System:**
-
-- `Microgrid`, `DeliveryArea`, and `Location` data classes
-- Protobuf integration with proper type safety
-
-**Exception Handling:**
-
-- Custom exception hierarchy (`AssetsApiError`, `NotFoundError`, `AuthenticationError`, `ServiceUnavailableError`)
-- JSON serialization support for error responses
+* **Type System**: Enhanced data classes with protobuf integration
+  * `Microgrid`, `DeliveryArea`, `Location`, and comprehensive electrical component types
+  * Proper enum mapping: `BatteryType`, `EvChargerType`, `InverterType`, `Metric`
 
 ## Bug Fixes
-
-- Improved dependency management with optional dependency groups
-- Enhanced gRPC error handling and type safety
-- Cleaned up deprecated code

src/frequenz/client/assets/__init__.py

@@ -4,5 +4,15 @@
 """Assets API client."""
 
 from ._client import AssetsApiClient
+from ._delivery_area import DeliveryArea, EnergyMarketCodeType
+from ._location import Location
+from ._microgrid import Microgrid, MicrogridStatus
 
-__all__ = ["AssetsApiClient"]
+__all__ = [
+    "AssetsApiClient",
+    "DeliveryArea",
+    "EnergyMarketCodeType",
+    "Microgrid",
+    "MicrogridStatus",
+    "Location",
+]

src/frequenz/client/assets/_client.py

@@ -10,36 +10,60 @@
 from __future__ import annotations
 
 from frequenz.api.assets.v1 import assets_pb2, assets_pb2_grpc
+from frequenz.client.base import channel
 from frequenz.client.base.client import BaseApiClient, call_stub_method
 
-from frequenz.client.assets.types import Microgrid
+from frequenz.client.assets.electrical_component._electrical_component import (
+    ElectricalComponent,
+)
 
+from ._microgrid import Microgrid
+from ._microgrid_proto import microgrid_from_proto
+from .electrical_component._electrical_component_proto import electrical_component_proto
 from .exceptions import ClientNotConnected
 
+DEFAULT_GRPC_CALL_TIMEOUT = 60.0
+"""The default timeout for gRPC calls made by this client (in seconds)."""
 
-class AssetsApiClient(BaseApiClient[assets_pb2_grpc.PlatformAssetsStub]):
+
+class AssetsApiClient(
+    BaseApiClient[assets_pb2_grpc.PlatformAssetsStub]
+):  # pylint: disable=too-many-arguments
     """A client for the Assets API."""
 
     def __init__(
         self,
         server_url: str,
-        auth_key: str | None,
-        sign_secret: str | None,
+        *,
+        auth_key: str | None = None,
+        sign_secret: str | None = None,
+        channel_defaults: channel.ChannelOptions = channel.ChannelOptions(),
         connect: bool = True,
     ) -> None:
         """
         Initialize the AssetsApiClient.
 
         Args:
-            server_url: The URL of the server to connect to.
-            auth_key: The API key to use when connecting to the service.
-            sign_secret: The secret to use when creating message HMAC.
-            connect: Whether to connect to the server as soon as a client instance is created.
+            server_url: The location of the microgrid API server in the form of a URL.
+                The following format is expected:
+                "grpc://hostname{:`port`}{?ssl=`ssl`}",
+                where the `port` should be an int between 0 and 65535 (defaulting to
+                9090) and `ssl` should be a boolean (defaulting to `true`).
+                For example: `grpc://localhost:1090?ssl=true`.
+            auth_key: The authentication key to use for the connection.
+            sign_secret: The secret to use for signing requests.
+            channel_defaults: The default options use to create the channel when not
+                specified in the URL.
+            connect: Whether to connect to the server as soon as a client instance is
+                created. If `False`, the client will not connect to the server until
+                [connect()][frequenz.client.base.client.BaseApiClient.connect] is
+                called.
         """
         super().__init__(
             server_url,
             assets_pb2_grpc.PlatformAssetsStub,
             connect=connect,
+            channel_defaults=channel_defaults,
             auth_key=auth_key,
             sign_secret=sign_secret,
         )
@@ -61,7 +85,7 @@ def stub(self) -> assets_pb2_grpc.PlatformAssetsAsyncStub:
         # use the async stub, so we cast the sync stub to the async stub.
         return self._stub  # type: ignore
 
-    async def get_microgrid_details(  # noqa: DOC502 (raises ApiClientError indirectly)
+    async def get_microgrid(  # noqa: DOC502 (raises ApiClientError indirectly)
         self, microgrid_id: int
     ) -> Microgrid:
         """
@@ -77,11 +101,40 @@ async def get_microgrid_details(  # noqa: DOC502 (raises ApiClientError indirect
             ApiClientError: If there are any errors communicating with the Assets API,
                 most likely a subclass of [GrpcError][frequenz.client.base.exception.GrpcError].
         """
-        request = assets_pb2.GetMicrogridRequest(microgrid_id=microgrid_id)
         response = await call_stub_method(
             self,
-            lambda: self.stub.GetMicrogrid(request),
+            lambda: self.stub.GetMicrogrid(
+                assets_pb2.GetMicrogridRequest(microgrid_id=microgrid_id),
+                timeout=DEFAULT_GRPC_CALL_TIMEOUT,
+            ),
             method_name="GetMicrogrid",
         )
 
-        return Microgrid.from_protobuf(response.microgrid)
+        return microgrid_from_proto(response.microgrid)
+
+    async def list_microgrid_electrical_components(
+        self, microgrid_id: int
+    ) -> list[ElectricalComponent]:
+        """
+        Get the electrical components of a microgrid.
+
+        Args:
+            microgrid_id: The ID of the microgrid to get the electrical components of.
+
+        Returns:
+            The electrical components of the microgrid.
+        """
+        response = await call_stub_method(
+            self,
+            lambda: self.stub.ListMicrogridElectricalComponents(
+                assets_pb2.ListMicrogridElectricalComponentsRequest(
+                    microgrid_id=microgrid_id,
+                ),
+                timeout=DEFAULT_GRPC_CALL_TIMEOUT,
+            ),
+            method_name="ListMicrogridElectricalComponents",
+        )
+
+        return [
+            electrical_component_proto(component) for component in response.components
+        ]

src/frequenz/client/assets/_utils.py

@@ -42,22 +42,22 @@ def default(self, o: Any) -> Any:
 
         return super().default(o)
 
-    def _encode(self, o: Any) -> Any:
+    def _encode_containers_recursively(self, o: Any) -> Any:
         """Recursively process objects to convert enum keys to their values."""
         if isinstance(o, dict):
             return {
                 (
-                    self._encode(key.value)
+                    self._encode_containers_recursively(key.value)
                     if isinstance(key, enum.Enum)
-                    else self._encode(key)
-                ): self._encode(value)
+                    else self._encode_containers_recursively(key)
+                ): self._encode_containers_recursively(value)
                 for key, value in o.items()
             }
-        if isinstance(o, (list, tuple, set, frozenset)):
-            items = [self._encode(item) for item in o]
+        elif isinstance(o, (list, tuple, set, frozenset)):
+            items = [self._encode_containers_recursively(item) for item in o]
             return items if isinstance(o, (list, tuple)) else items
         return o
 
     def encode(self, o: Any) -> str:
         """Encode the given object to a JSON string, handling enum keys."""
-        return super().encode(self._encode(o))
+        return super().encode(self._encode_containers_recursively(o))

src/frequenz/client/assets/cli/__main__.py

@@ -34,33 +34,13 @@
 
 import asyncio
 import json
-from dataclasses import asdict
 
 import asyncclick as click
 
 from frequenz.client.assets._client import AssetsApiClient
 from frequenz.client.assets.exceptions import ApiClientError
-from frequenz.client.assets.types import Microgrid
 
-
-def print_microgrid_details(microgrid: Microgrid) -> None:
-    """
-    Print microgrid details to console in JSON format.
-
-    This function converts the Microgrid instance to a dictionary and
-    outputs it as formatted JSON to the console. The output is designed
-    to be machine-readable and can be piped to tools like jq for further
-    processing.
-
-    Args:
-        microgrid: The Microgrid instance to print to console.
-    """
-    microgrid_dict = asdict(microgrid)
-    microgrid_dict["id"] = int(microgrid.id)
-    microgrid_dict["enterprise_id"] = int(microgrid.enterprise_id)
-    microgrid_dict["create_time"] = microgrid.create_time.isoformat()
-
-    click.echo(json.dumps(microgrid_dict, indent=2))
+from ._utils import print_electrical_components, print_microgrid_details
 
 
 @click.group(invoke_without_command=True)
@@ -161,7 +141,7 @@ async def get_microgrid(
     """
     try:
         client = ctx.obj["client"]
-        microgrid_details = await client.get_microgrid_details(microgrid_id)
+        microgrid_details = await client.get_microgrid(microgrid_id)
         print_microgrid_details(microgrid_details)
     except ApiClientError as e:
         error_dict = {
@@ -174,6 +154,53 @@ async def get_microgrid(
         raise click.Abort()
 
 
+@cli.command("electrical-components")
+@click.pass_context
+@click.argument("microgrid-id", required=True, type=int)
+async def list_microgrid_electrical_components(
+    ctx: click.Context,
+    microgrid_id: int,
+) -> None:
+    """
+    Get and display electrical components by microgrid ID.
+
+    This command fetches detailed information about all electrical components
+    in a specific microgrid from the Assets API and displays it in JSON format.
+    The output can be piped to other tools for further processing.
+
+    Args:
+        ctx: Click context object containing the initialized API client.
+        microgrid_id: The unique identifier of the microgrid to retrieve.
+
+    Raises:
+        click.Abort: If there is an error printing the electrical components.
+
+    Example:
+        ```bash
+        # Get details for microgrid with ID 123
+        assets-cli electrical-components 123
+
+        # Pipe output to jq for filtering
+        assets-cli electrical-components 123 | jq ".id"
+        ```
+    """
+    try:
+        client = ctx.obj["client"]
+        electrical_components = await client.list_microgrid_electrical_components(
+            microgrid_id
+        )
+        print_electrical_components(electrical_components)
+    except ApiClientError as e:
+        error_dict = {
+            "error_type": type(e).__name__,
+            "server_url": e.server_url,
+            "operation": e.operation,
+            "description": e.description,
+        }
+        click.echo(json.dumps(error_dict, indent=2))
+        raise click.Abort()
+
+
 def main() -> None:
     """
     Initialize and run the CLI application.

src/frequenz/client/assets/exceptions.py

@@ -47,4 +47,5 @@
     "ServiceUnavailable",
     "UnknownError",
     "UnrecognizedGrpcStatus",
+    "PermissionDenied",
 ]