Add timestamp <-> datetime conversion functions

These functions are copied from `frequenz-client-base` and renamed to
match the naming convention in this project. The functions in
`frequenz-client-base` will be eventually deprecated.

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

Author: llucax

pyproject.toml

@@ -29,6 +29,7 @@ dependencies = [
   "typing-extensions >= 4.13.0, < 5",
   "frequenz-api-common >= 0.8.0, < 1",
   "frequenz-core >= 1.0.2, < 2",
+  "protobuf >= 6.33.1, < 7",
 ]
 dynamic = ["version"]
 
@@ -60,6 +61,7 @@ dev-mkdocs = [
 dev-mypy = [
   "mypy == 1.18.2",
   "types-Markdown == 3.9.0.20250906",
+  "types-protobuf == 6.32.1.20251105",
   # For checking the noxfile, docs/ script, and tests
   "frequenz-client-common[dev-mkdocs,dev-noxfile,dev-pytest]",
 ]
@@ -72,6 +74,7 @@ dev-pylint = [
 dev-pytest = [
   "pytest == 8.4.2",
   "frequenz-repo-config[extra-lint-examples] == 0.13.6",
+  "hypothesis == 6.140.3",
   "pytest-mock == 3.15.1",
   "pytest-asyncio == 1.2.0",
   "async-solipsism == 0.8",

src/frequenz/client/common/proto/__init__.py

@@ -4,7 +4,10 @@
 """General utilities for converting common types to/from protobuf types."""
 
 from ._enum import enum_from_proto
+from ._timestamp import datetime_from_proto, datetime_to_proto
 
 __all__ = [
     "enum_from_proto",
+    "datetime_from_proto",
+    "datetime_to_proto",
 ]

src/frequenz/client/common/proto/_timestamp.py

@@ -0,0 +1,69 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Helper functions to convert protobuf Timestamp <-> Python datetime."""
+
+from datetime import datetime, timezone
+from typing import overload
+
+from google.protobuf import timestamp_pb2
+
+
+@overload
+def datetime_to_proto(dt: datetime) -> timestamp_pb2.Timestamp:
+    """Convert a datetime to a protobuf Timestamp.
+
+    Args:
+        dt: datetime object to convert
+
+    Returns:
+        datetime converted to Timestamp
+    """
+
+
+@overload
+def datetime_to_proto(dt: None) -> None:
+    """Overload to handle None values.
+
+    Args:
+        dt: None
+
+    Returns:
+        None
+    """
+
+
+def datetime_to_proto(dt: datetime | None) -> timestamp_pb2.Timestamp | None:
+    """Convert a datetime to a protobuf Timestamp.
+
+    Returns None if dt is None.
+
+    Args:
+        dt: datetime object to convert
+
+    Returns:
+        datetime converted to Timestamp
+    """
+    if dt is None:
+        return None
+
+    ts = timestamp_pb2.Timestamp()
+    ts.FromDatetime(dt)
+    return ts
+
+
+def datetime_from_proto(
+    ts: timestamp_pb2.Timestamp, tz: timezone = timezone.utc
+) -> datetime:
+    """Convert a protobuf Timestamp to a datetime.
+
+    Args:
+        ts: Timestamp object to convert
+        tz: Timezone to use for the datetime
+
+    Returns:
+        Timestamp converted to datetime
+    """
+    # Add microseconds and add nanoseconds converted to microseconds
+    microseconds = int(ts.nanos / 1000)
+    return datetime.fromtimestamp(ts.seconds + microseconds * 1e-6, tz=tz)

tests/proto/test_datetime.py

@@ -0,0 +1,69 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Test conversion helper functions."""
+
+from datetime import datetime, timezone
+
+# pylint: disable=no-name-in-module
+from google.protobuf.timestamp_pb2 import Timestamp
+
+# pylint: enable=no-name-in-module
+from hypothesis import given
+from hypothesis import strategies as st
+
+from frequenz.client.common.proto import datetime_from_proto, datetime_to_proto
+
+# Strategy for generating datetime objects
+datetime_strategy = st.datetimes(
+    min_value=datetime(1970, 1, 1),
+    max_value=datetime(9999, 12, 31),
+    timezones=st.just(timezone.utc),
+)
+
+# Strategy for generating Timestamp objects
+timestamp_strategy = st.builds(
+    Timestamp,
+    seconds=st.integers(
+        min_value=0,
+        max_value=int(datetime(9999, 12, 31, tzinfo=timezone.utc).timestamp()),
+    ),
+)
+
+
+@given(datetime_strategy)
+def test_to_timestamp_with_datetime(dt: datetime) -> None:
+    """Test conversion from datetime to Timestamp."""
+    ts = datetime_to_proto(dt)
+    assert ts is not None
+    converted_back_dt = datetime_from_proto(ts)
+    assert dt.tzinfo == converted_back_dt.tzinfo
+    assert dt.timestamp() == converted_back_dt.timestamp()
+
+
+def test_to_timestamp_with_none() -> None:
+    """Test that passing None returns None."""
+    assert datetime_to_proto(None) is None
+
+
+@given(timestamp_strategy)
+def test_to_datetime(ts: Timestamp) -> None:
+    """Test conversion from Timestamp to datetime."""
+    dt = datetime_from_proto(ts)
+    assert dt is not None
+    # Convert back to Timestamp and compare
+    converted_back_ts = datetime_to_proto(dt)
+    assert ts.seconds == converted_back_ts.seconds
+
+
+@given(datetime_strategy)
+def test_no_none_datetime(dt: datetime) -> None:
+    """Test behavior of type hinting."""
+    ts: Timestamp = datetime_to_proto(dt)
+    dt_none: datetime | None = None
+
+    # The test would fail without the ignore comment as it should.
+    ts2: Timestamp = datetime_to_proto(dt_none)  # type: ignore
+
+    assert ts is not None
+    assert ts2 is None