-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add datetime and Timestamp conversion functions
Signed-off-by: Mathias L. Baumann <[email protected]>
- Loading branch information
Showing
4 changed files
with
142 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,68 @@ | ||
# License: MIT | ||
# Copyright © 2024 Frequenz Energy-as-a-Service GmbH | ||
|
||
"""Helper functions to convert to/from common python types.""" | ||
|
||
from datetime import datetime, timezone | ||
from typing import overload | ||
|
||
# pylint: disable=no-name-in-module | ||
from google.protobuf.timestamp_pb2 import Timestamp | ||
|
||
# pylint: enable=no-name-in-module | ||
|
||
|
||
@overload | ||
def to_timestamp(dt: datetime) -> Timestamp: | ||
"""Convert a datetime to a protobuf Timestamp. | ||
Args: | ||
dt: datetime object to convert | ||
Returns: | ||
datetime converted to Timestamp | ||
""" | ||
|
||
|
||
@overload | ||
def to_timestamp(dt: None) -> None: | ||
"""Overload to handle None values. | ||
Args: | ||
dt: None | ||
Returns: | ||
None | ||
""" | ||
|
||
|
||
def to_timestamp(dt: datetime | None) -> 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() | ||
ts.FromDatetime(dt) | ||
return ts | ||
|
||
|
||
def to_datetime(ts: 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 | ||
""" | ||
return datetime.fromtimestamp(ts.seconds, tz=tz) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,71 @@ | ||
# License: MIT | ||
# Copyright © 2024 Frequenz Energy-as-a-Service GmbH | ||
|
||
"""Test conversion helper functions.""" | ||
|
||
from datetime import datetime, timezone | ||
from math import isclose | ||
|
||
# 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.base.conversion_helper import to_datetime, to_timestamp | ||
|
||
# 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 = to_timestamp(dt) | ||
assert ts is not None | ||
converted_back_dt = to_datetime(ts) | ||
# Compare timestamps to account for floating point errors | ||
# Up to 1 second difference is allowed | ||
assert isclose(dt.timestamp(), converted_back_dt.timestamp(), abs_tol=1) | ||
|
||
|
||
def test_to_timestamp_with_none() -> None: | ||
"""Test that passing None returns None.""" | ||
assert to_timestamp(None) is None | ||
|
||
|
||
@given(timestamp_strategy) | ||
def test_to_datetime(ts: Timestamp) -> None: | ||
"""Test conversion from Timestamp to datetime.""" | ||
dt = to_datetime(ts) | ||
assert dt is not None | ||
# Convert back to Timestamp and compare | ||
converted_back_ts = to_timestamp(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 = to_timestamp(dt) | ||
dt_none: datetime | None = None | ||
|
||
# The test would fail without the ignore comment as it should. | ||
ts2: Timestamp = to_timestamp(dt_none) # type: ignore | ||
|
||
assert ts is not None | ||
assert ts2 is None |