Add loading of Lifetime from protobuf

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

Author: llucax

src/frequenz/client/microgrid/_lifetime_proto.py

@@ -0,0 +1,26 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Loading of Lifetime objects from protobuf messages."""
+
+from frequenz.api.common.v1.microgrid import lifetime_pb2
+from frequenz.client.base.conversion import to_datetime
+
+from ._lifetime import Lifetime
+
+
+def lifetime_from_proto(
+    message: lifetime_pb2.Lifetime,
+) -> Lifetime:
+    """Create a [`Lifetime`][frequenz.client.microgrid.Lifetime] from a protobuf message."""
+    start = (
+        to_datetime(message.start_timestamp)
+        if message.HasField("start_timestamp")
+        else None
+    )
+    end = (
+        to_datetime(message.end_timestamp)
+        if message.HasField("end_timestamp")
+        else None
+    )
+    return Lifetime(start=start, end=end)

tests/test_lifetime.py

@@ -1,15 +1,19 @@
 # License: MIT
 # Copyright © 2025 Frequenz Energy-as-a-Service GmbH
 
-"""Tests for the Lifetime class."""
+"""Tests for the Lifetime class and its protobuf conversion."""
 
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from enum import Enum, auto
+from typing import Any
 
 import pytest
+from frequenz.api.common.v1.microgrid import lifetime_pb2
+from google.protobuf import timestamp_pb2
 
 from frequenz.client.microgrid import Lifetime
+from frequenz.client.microgrid._lifetime_proto import lifetime_from_proto
 
 
 class _Time(Enum):
@@ -65,6 +69,20 @@ class _ActivityTestCase:
     """The expected operational state."""
 
 
+@dataclass(frozen=True, kw_only=True)
+class _ProtoConversionTestCase:
+    """Test case for protobuf conversion."""
+
+    name: str
+    """The description of the test case."""
+
+    include_start: bool
+    """Whether to include start timestamp."""
+
+    include_end: bool
+    """Whether to include end timestamp."""
+
+
 @dataclass(frozen=True, kw_only=True)
 class _FixedLifetimeTestCase:
     """Test case for fixed lifetime activity testing."""
@@ -303,3 +321,51 @@ def test_active_at_with_fixed_lifetime(
     }[case.test_time]
 
     assert lifetime.is_operational_at(test_time) == case.expected_operational
+
+
+@pytest.mark.parametrize(
+    "case",
+    [
+        _ProtoConversionTestCase(
+            name="both timestamps", include_start=True, include_end=True
+        ),
+        _ProtoConversionTestCase(
+            name="only start timestamp", include_start=True, include_end=False
+        ),
+        _ProtoConversionTestCase(
+            name="only end timestamp", include_start=False, include_end=True
+        ),
+        _ProtoConversionTestCase(
+            name="no timestamps", include_start=False, include_end=False
+        ),
+    ],
+    ids=lambda case: case.name,
+)
+def test_from_proto(
+    now: datetime, future: datetime, case: _ProtoConversionTestCase
+) -> None:
+    """Test conversion from protobuf message to Lifetime."""
+    now_ts = timestamp_pb2.Timestamp()
+    now_ts.FromDatetime(now)
+
+    future_ts = timestamp_pb2.Timestamp()
+    future_ts.FromDatetime(future)
+
+    proto_kwargs: dict[str, Any] = {}
+    if case.include_start:
+        proto_kwargs["start_timestamp"] = now_ts
+    if case.include_end:
+        proto_kwargs["end_timestamp"] = future_ts
+
+    proto = lifetime_pb2.Lifetime(**proto_kwargs)
+    lifetime = lifetime_from_proto(proto)
+
+    if case.include_start:
+        assert lifetime.start == now
+    else:
+        assert lifetime.start is None
+
+    if case.include_end:
+        assert lifetime.end == future
+    else:
+        assert lifetime.end is None