forked from GoogleCloudPlatform/ml-auto-solutions
-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathtime_util.py
More file actions
90 lines (67 loc) · 2.98 KB
/
time_util.py
File metadata and controls
90 lines (67 loc) · 2.98 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
"""Utility class for handling various time representations."""
from __future__ import annotations
import datetime
from dataclasses import dataclass
from google.protobuf import timestamp_pb2
@dataclass
class TimeUtil:
"""A utility that represents time in multiple forms and converting between them."""
time: int
@classmethod
def now(cls) -> TimeUtil:
"""Returns the current time in UTC."""
return cls(int(datetime.datetime.now(datetime.timezone.utc).timestamp()))
@classmethod
def from_iso_string(cls, time_str: str) -> TimeUtil:
"""Builds a TimeUtil object from an ISO 8601 formatted string."""
dt_object = datetime.datetime.fromisoformat(time_str.replace("Z", "+00:00"))
return cls(int(dt_object.timestamp()))
@classmethod
def from_timestamp_pb2(cls, ts_pb: timestamp_pb2.Timestamp) -> TimeUtil:
"""Builds a TimeUtil object from a Google Protobuf Timestamp."""
return cls(int(ts_pb.seconds))
@classmethod
def from_datetime(cls, dt: datetime.datetime) -> TimeUtil:
"""Builds a TimeUtil object from a standard datetime object."""
return cls(int(dt.timestamp()))
@classmethod
def from_unix_seconds(cls, unix_seconds: int | float) -> TimeUtil:
"""Builds a TimeUtil object from a Unix timestamp (seconds)."""
return cls(int(unix_seconds))
def to_unix_seconds(self) -> int:
return self.time
def to_timestamp_pb2(self) -> timestamp_pb2.Timestamp:
timestamp = timestamp_pb2.Timestamp()
timestamp.FromSeconds(self.time)
return timestamp
def to_datetime(self) -> datetime.datetime:
return datetime.datetime.fromtimestamp(self.time, tz=datetime.timezone.utc)
def to_iso_string(self) -> str:
iso_str = self.to_datetime().isoformat()
return iso_str.replace("+00:00", "Z")
def to_mql_string(self) -> str:
dt = self.to_datetime()
return dt.strftime("d'%Y/%m/%d-%H:%M:%S'")
def __add__(self, other: datetime.timedelta) -> TimeUtil:
"""Allows usage like: TimeUtil(...) + timedelta(minutes=10)."""
if isinstance(other, datetime.timedelta):
return TimeUtil(self.time + int(other.total_seconds()))
return NotImplemented
def __sub__(self, other: datetime.timedelta | TimeUtil) -> TimeUtil:
"""Allows usage like: TimeUtil(...) - timedelta(minutes=10) or TimeUtil(...) - TimeUtil(...)."""
if isinstance(other, datetime.timedelta):
return TimeUtil(self.time - int(other.total_seconds()))
if isinstance(other, TimeUtil):
return TimeUtil(self.time - other.time)
return NotImplemented
if __name__ == "__main__":
time = "2025-09-19T04:08:35.951+00:00"
time_obj = TimeUtil.from_iso_string(time)
print(time_obj.to_iso_string())
print(time_obj.to_timestamp_pb2())
start_time = datetime.datetime.fromisoformat("2025-09-19T04:08:35.951+00:00")
end_time = start_time + datetime.timedelta(minutes=10)
start_time_obj = TimeUtil.from_datetime(start_time)
end_time_obj = TimeUtil.from_datetime(end_time)
print(start_time_obj.to_timestamp_pb2())
print(end_time_obj.to_timestamp_pb2())