python (3.11.7)
1 # coding: utf-8
2 from collections import namedtuple
3 import datetime
4 import sys
5 import struct
6
7
8 PY2 = sys.version_info[0] == 2
9
10 if PY2:
11 int_types = (int, long)
12 _utc = None
13 else:
14 int_types = int
15 try:
16 _utc = datetime.timezone.utc
17 except AttributeError:
18 _utc = datetime.timezone(datetime.timedelta(0))
19
20
21 class ESC[4;38;5;81mExtType(ESC[4;38;5;149mnamedtuple("ExtType", "code data")):
22 """ExtType represents ext type in msgpack."""
23
24 def __new__(cls, code, data):
25 if not isinstance(code, int):
26 raise TypeError("code must be int")
27 if not isinstance(data, bytes):
28 raise TypeError("data must be bytes")
29 if not 0 <= code <= 127:
30 raise ValueError("code must be 0~127")
31 return super(ExtType, cls).__new__(cls, code, data)
32
33
34 class ESC[4;38;5;81mTimestamp(ESC[4;38;5;149mobject):
35 """Timestamp represents the Timestamp extension type in msgpack.
36
37 When built with Cython, msgpack uses C methods to pack and unpack `Timestamp`. When using pure-Python
38 msgpack, :func:`to_bytes` and :func:`from_bytes` are used to pack and unpack `Timestamp`.
39
40 This class is immutable: Do not override seconds and nanoseconds.
41 """
42
43 __slots__ = ["seconds", "nanoseconds"]
44
45 def __init__(self, seconds, nanoseconds=0):
46 """Initialize a Timestamp object.
47
48 :param int seconds:
49 Number of seconds since the UNIX epoch (00:00:00 UTC Jan 1 1970, minus leap seconds).
50 May be negative.
51
52 :param int nanoseconds:
53 Number of nanoseconds to add to `seconds` to get fractional time.
54 Maximum is 999_999_999. Default is 0.
55
56 Note: Negative times (before the UNIX epoch) are represented as negative seconds + positive ns.
57 """
58 if not isinstance(seconds, int_types):
59 raise TypeError("seconds must be an integer")
60 if not isinstance(nanoseconds, int_types):
61 raise TypeError("nanoseconds must be an integer")
62 if not (0 <= nanoseconds < 10**9):
63 raise ValueError(
64 "nanoseconds must be a non-negative integer less than 999999999."
65 )
66 self.seconds = seconds
67 self.nanoseconds = nanoseconds
68
69 def __repr__(self):
70 """String representation of Timestamp."""
71 return "Timestamp(seconds={0}, nanoseconds={1})".format(
72 self.seconds, self.nanoseconds
73 )
74
75 def __eq__(self, other):
76 """Check for equality with another Timestamp object"""
77 if type(other) is self.__class__:
78 return (
79 self.seconds == other.seconds and self.nanoseconds == other.nanoseconds
80 )
81 return False
82
83 def __ne__(self, other):
84 """not-equals method (see :func:`__eq__()`)"""
85 return not self.__eq__(other)
86
87 def __hash__(self):
88 return hash((self.seconds, self.nanoseconds))
89
90 @staticmethod
91 def from_bytes(b):
92 """Unpack bytes into a `Timestamp` object.
93
94 Used for pure-Python msgpack unpacking.
95
96 :param b: Payload from msgpack ext message with code -1
97 :type b: bytes
98
99 :returns: Timestamp object unpacked from msgpack ext payload
100 :rtype: Timestamp
101 """
102 if len(b) == 4:
103 seconds = struct.unpack("!L", b)[0]
104 nanoseconds = 0
105 elif len(b) == 8:
106 data64 = struct.unpack("!Q", b)[0]
107 seconds = data64 & 0x00000003FFFFFFFF
108 nanoseconds = data64 >> 34
109 elif len(b) == 12:
110 nanoseconds, seconds = struct.unpack("!Iq", b)
111 else:
112 raise ValueError(
113 "Timestamp type can only be created from 32, 64, or 96-bit byte objects"
114 )
115 return Timestamp(seconds, nanoseconds)
116
117 def to_bytes(self):
118 """Pack this Timestamp object into bytes.
119
120 Used for pure-Python msgpack packing.
121
122 :returns data: Payload for EXT message with code -1 (timestamp type)
123 :rtype: bytes
124 """
125 if (self.seconds >> 34) == 0: # seconds is non-negative and fits in 34 bits
126 data64 = self.nanoseconds << 34 | self.seconds
127 if data64 & 0xFFFFFFFF00000000 == 0:
128 # nanoseconds is zero and seconds < 2**32, so timestamp 32
129 data = struct.pack("!L", data64)
130 else:
131 # timestamp 64
132 data = struct.pack("!Q", data64)
133 else:
134 # timestamp 96
135 data = struct.pack("!Iq", self.nanoseconds, self.seconds)
136 return data
137
138 @staticmethod
139 def from_unix(unix_sec):
140 """Create a Timestamp from posix timestamp in seconds.
141
142 :param unix_float: Posix timestamp in seconds.
143 :type unix_float: int or float.
144 """
145 seconds = int(unix_sec // 1)
146 nanoseconds = int((unix_sec % 1) * 10**9)
147 return Timestamp(seconds, nanoseconds)
148
149 def to_unix(self):
150 """Get the timestamp as a floating-point value.
151
152 :returns: posix timestamp
153 :rtype: float
154 """
155 return self.seconds + self.nanoseconds / 1e9
156
157 @staticmethod
158 def from_unix_nano(unix_ns):
159 """Create a Timestamp from posix timestamp in nanoseconds.
160
161 :param int unix_ns: Posix timestamp in nanoseconds.
162 :rtype: Timestamp
163 """
164 return Timestamp(*divmod(unix_ns, 10**9))
165
166 def to_unix_nano(self):
167 """Get the timestamp as a unixtime in nanoseconds.
168
169 :returns: posix timestamp in nanoseconds
170 :rtype: int
171 """
172 return self.seconds * 10**9 + self.nanoseconds
173
174 def to_datetime(self):
175 """Get the timestamp as a UTC datetime.
176
177 Python 2 is not supported.
178
179 :rtype: datetime.
180 """
181 return datetime.datetime.fromtimestamp(0, _utc) + datetime.timedelta(
182 seconds=self.to_unix()
183 )
184
185 @staticmethod
186 def from_datetime(dt):
187 """Create a Timestamp from datetime with tzinfo.
188
189 Python 2 is not supported.
190
191 :rtype: Timestamp
192 """
193 return Timestamp.from_unix(dt.timestamp())
Browse
Usage
Versions