Add loading of AggregatedMetricValue from protobuf

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

Author: llucax

src/frequenz/client/microgrid/metrics/_sample_proto.py

@@ -0,0 +1,27 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Loading of MetricSample and AggregatedMetricValue objects from protobuf messages."""
+
+from frequenz.api.common.v1.metrics import metric_sample_pb2
+
+from ._sample import AggregatedMetricValue
+
+
+def aggregated_metric_sample_from_proto(
+    message: metric_sample_pb2.AggregatedMetricValue,
+) -> AggregatedMetricValue:
+    """Convert a protobuf message to a `AggregatedMetricValue` object.
+
+    Args:
+        message: The protobuf message to convert.
+
+    Returns:
+        The resulting `AggregatedMetricValue` object.
+    """
+    return AggregatedMetricValue(
+        avg=message.avg_value,
+        min=message.min_value if message.HasField("min_value") else None,
+        max=message.max_value if message.HasField("max_value") else None,
+        raw_values=message.raw_values,
+    )

tests/metrics/test_sample.py

@@ -3,7 +3,41 @@
 
 """Tests for the Sample class and related classes."""
 
+from dataclasses import dataclass, field
+
+import pytest
+from frequenz.api.common.v1.metrics import metric_sample_pb2
+
 from frequenz.client.microgrid.metrics import AggregatedMetricValue, AggregationMethod
+from frequenz.client.microgrid.metrics._sample_proto import (
+    aggregated_metric_sample_from_proto,
+)
+
+
+@dataclass(frozen=True, kw_only=True)
+class _AggregatedValueTestCase:
+    """Test case for AggregatedMetricValue protobuf conversion."""
+
+    name: str
+    """The description of the test case."""
+
+    avg_value: float
+    """The average value to set."""
+
+    has_min: bool = True
+    """Whether to include min value."""
+
+    has_max: bool = True
+    """Whether to include max value."""
+
+    min_value: float | None = None
+    """The minimum value to set."""
+
+    max_value: float | None = None
+    """The maximum value to set."""
+
+    raw_values: list[float] = field(default_factory=list)
+    """The raw values to include."""
 
 
 def test_aggregation_method_values() -> None:
@@ -40,3 +74,54 @@ def test_aggregated_metric_value() -> None:
     assert value.max is None
     assert not value.raw_values
     assert str(value) == "avg:5.0"
+
+
+@pytest.mark.parametrize(
+    "case",
+    [
+        _AggregatedValueTestCase(
+            name="full",
+            avg_value=5.0,
+            min_value=1.0,
+            max_value=10.0,
+            raw_values=[1.0, 5.0, 10.0],
+        ),
+        _AggregatedValueTestCase(
+            name="minimal",
+            avg_value=5.0,
+            has_min=False,
+            has_max=False,
+        ),
+        _AggregatedValueTestCase(
+            name="only_min",
+            avg_value=5.0,
+            has_max=False,
+            min_value=1.0,
+        ),
+        _AggregatedValueTestCase(
+            name="only_max",
+            avg_value=5.0,
+            has_min=False,
+            max_value=10.0,
+        ),
+    ],
+    ids=lambda case: case.name,
+)
+def test_aggregated_metric_value_from_proto(case: _AggregatedValueTestCase) -> None:
+    """Test conversion from protobuf message to AggregatedMetricValue."""
+    proto = metric_sample_pb2.AggregatedMetricValue(
+        avg_value=case.avg_value,
+    )
+    if case.has_min and case.min_value is not None:
+        proto.min_value = case.min_value
+    if case.has_max and case.max_value is not None:
+        proto.max_value = case.max_value
+    if case.raw_values:
+        proto.raw_values.extend(case.raw_values)
+
+    value = aggregated_metric_sample_from_proto(proto)
+
+    assert value.avg == case.avg_value
+    assert value.min == (case.min_value if case.has_min else None)
+    assert value.max == (case.max_value if case.has_max else None)
+    assert list(value.raw_values) == case.raw_values