edq.util.time

  1import datetime
  2import re
  3import time
  4import typing
  5
  6import edq.util.serial
  7
  8PRETTY_SHORT_FORMAT: str = '%Y-%m-%d %H:%M'
  9"""
 10The format string for a pretty timestamp.
 11See: https://docs.python.org/3/library/datetime.html#strftime-strptime-behavior
 12"""
 13
 14DEFAULT_EMBEDDED_PATTERN: str = r'<timestamp:(-?\d+|nil)>'
 15""" A regex for matching an embedded timestamp. """
 16
 17UTC: datetime.timezone = datetime.timezone.utc
 18""" A shortcut for the UTC timezone. """
 19
 20UNIXTIME_THRESHOLD_SECS: int = int(1e10)
 21""" Epoch time guessing threshold for seconds. """
 22
 23UNIXTIME_THRESHOLD_MSECS: int = int(1e13)
 24""" Epoch time guessing threshold for milliseconds. """
 25
 26UNIXTIME_THRESHOLD_USECS: int = int(1e16)
 27""" Epoch time guessing threshold for nanoseconds. """
 28
 29_testing_timezone: typing.Union[datetime.tzinfo, None] = None  # pylint: disable=invalid-name
 30""" A timezone to use for testing. """
 31
 32def set_testing_local_timezone(timezone: typing.Union[datetime.tzinfo, None] = UTC) -> None:
 33    """
 34    Force the local timezone to be a specific value (UTC by default).
 35    This will only affect this package (e.g., the stdlib will not be affected).
 36    """
 37
 38    global _testing_timezone  # pylint: disable=global-statement
 39    _testing_timezone = timezone
 40
 41class Duration(int):
 42    """
 43    A Duration represents some length in time in milliseconds.
 44    """
 45
 46    def to_secs(self) -> float:
 47        """ Convert the duration to float seconds. """
 48
 49        return self / 1000.0
 50
 51    def to_msecs(self) -> int:
 52        """ Convert the duration to integer milliseconds. """
 53
 54        return self
 55
 56class Timestamp(int, edq.util.serial.PODConverter):  # type: ignore[misc]
 57    """
 58    A Timestamp represent a moment in time (sometimes called "datetimes").
 59    Timestamps are internally represented by the number of milliseconds since the
 60    (Unix Epoch)[https://en.wikipedia.org/wiki/Unix_time].
 61    This is sometimes referred to as "Unix Time".
 62    Since Unix Time is in UTC, timestamps do not need to carry timestamp information with them.
 63
 64    Note that timestamps are just integers with some decoration,
 65    so they respond to all normal int functionality.
 66    """
 67
 68    def sub(self, other: 'Timestamp') -> Duration:
 69        """ Return a new duration that is the difference of this and the given duration. """
 70
 71        return Duration(self - other)
 72
 73    def to_pytime(self, timezone: typing.Union[datetime.tzinfo, None] = None) -> datetime.datetime:
 74        """ Convert this timestamp to a Python datetime in the given timezone (local by default). """
 75
 76        if (timezone is None):
 77            timezone = get_local_timezone()
 78
 79        return datetime.datetime.fromtimestamp(self / 1000, timezone)
 80
 81    def to_local_pytime(self) -> datetime.datetime:
 82        """ Convert this timestamp to a Python datetime in the system timezone. """
 83
 84        return self.to_pytime(timezone = get_local_timezone())
 85
 86    def pretty(self, short: bool = False, timezone: typing.Union[datetime.tzinfo, None] = None) -> str:
 87        """
 88        Get a "pretty" string representation of this timestamp.
 89        There is no guarantee that this representation can be parsed back to its original form.
 90
 91        If no timezone is provided, the system's local timezone will be used.
 92        """
 93
 94        if (timezone is None):
 95            timezone = get_local_timezone()
 96
 97        pytime = self.to_pytime(timezone = timezone)
 98
 99        if (short):
100            return pytime.strftime(PRETTY_SHORT_FORMAT)
101
102        return pytime.isoformat(timespec = 'milliseconds')
103
104    def strftime(self, format_string: str, timezone: typing.Union[datetime.tzinfo, None] = None) -> str:
105        """
106        Convert the timesstamp to Python datetime and then call strftime with the given format.
107        """
108
109        if (timezone is None):
110            timezone = get_local_timezone()
111
112        pytime = self.to_pytime(timezone = timezone)
113
114        return pytime.strftime(format_string)
115
116    @staticmethod
117    def from_pytime(pytime: datetime.datetime) -> 'Timestamp':
118        """ Convert a Python datetime to a timestamp. """
119
120        return Timestamp(int(pytime.timestamp() * 1000))
121
122    @staticmethod
123    def now() -> 'Timestamp':
124        """ Get a Timestamp that represents the current moment. """
125
126        return Timestamp(time.time() * 1000)
127
128    @staticmethod
129    def convert_embedded(
130            text: str,
131            embedded_pattern: str = DEFAULT_EMBEDDED_PATTERN,
132            pretty: bool = False,
133            short: bool = True,
134            timezone: typing.Union[datetime.tzinfo, None] = None,
135            ) -> str:
136        """
137        Look for any timestamps embedded in the text and replace them.
138        """
139
140        while True:
141            match = re.search(embedded_pattern, text)
142            if (match is None):
143                break
144
145            initial_text = match.group(0)
146            timestamp_text = match.group(1)
147
148            timestamp = Timestamp()
149            if (timestamp_text != 'nil'):
150                timestamp = Timestamp(int(timestamp_text))
151
152            replacement_text = str(timestamp)
153            if (pretty):
154                replacement_text = timestamp.pretty(short = short, timezone = timezone)
155
156            text = text.replace(initial_text, replacement_text)
157
158        return text
159
160    @staticmethod
161    def guess(
162            value: typing.Any,
163            default_timezone: typing.Union[datetime.tzinfo, None] = None,
164            use_local_timezone: bool = True,
165            ) -> 'Timestamp':
166        """
167        Try to parse a timestamp out of a value.
168        Empty values will get zero timestamps.
169        Purely digit strings will be converted to ints and treated as UNIX times.
170        Floats will be considered UNIX epoch seconds and converted to milliseconds.
171        Other strings will be attempted to be parsed with datetime.fromisoformat().
172
173        If a value, but not a timezone, was parsed,
174        then set the timezone for the parsed value to `default_timezone`.
175        If `default_timezone` is None and `use_local_timezone` is true,
176        then use get_local_timezone() to get the timezone.
177        """
178
179        if ((default_timezone is None) and use_local_timezone):
180            default_timezone = get_local_timezone()
181
182        raw_value = value
183
184        # Empty timestamp.
185        if (value is None):
186            return Timestamp(0)
187
188        # Check for already parsed timestamps.
189        if (isinstance(value, Timestamp)):
190            return value
191
192        # Floats are assumed to be epoch seconds.
193        if (isinstance(value, float)):
194            value = int(1000 * value)
195
196        # At this point, we only want to be dealing with strings or ints.
197        if (not isinstance(value, (int, str))):
198            value = str(value)
199
200        # Check for string specifics.
201        if (isinstance(value, str)):
202            # Check for empty strings.
203            value = value.strip()
204            if (len(value) == 0):
205                return Timestamp(0)
206
207            # Check for digit or float strings.
208            if (re.match(r'^\d+\.\d+$', value) is not None):
209                value = int(1000 * float(value))
210            elif (re.match(r'^\d+$', value) is not None):
211                value = int(value)
212
213        if (isinstance(value, int)):
214		    # Use reasonable thresholds to guess the units of the value (sec, msec, usec, nsec).
215            if (value < UNIXTIME_THRESHOLD_SECS):
216                # Time is in seconds.
217                return Timestamp(value * 1000)
218            elif (value < UNIXTIME_THRESHOLD_MSECS):
219                # Time is in milliseconds.
220                return Timestamp(value)
221            elif (value < UNIXTIME_THRESHOLD_USECS):
222                # Time is in microseconds.
223                return Timestamp(value / 1000)
224            else:
225                # Time is in nanoseconds.
226                return Timestamp(value / 1000 / 1000)
227
228        # Try to convert from an ISO string.
229
230        # Parse out some cases that Python <= 3.10 cannot deal with.
231        # This will remove fractional seconds.
232        value = re.sub(r'Z$', '+00:00', value)
233        value = re.sub(r'(\d\d:\d\d)(\.\d+)', r'\1', value)
234
235        try:
236            value = datetime.datetime.fromisoformat(value)
237        except Exception as ex:
238            raise ValueError(f"Failed to parse timestamp string '{raw_value}'.") from ex
239
240        # If no timezone was parsed, use the default one.
241        if (value.tzinfo is None):
242            value = value.replace(tzinfo = default_timezone)
243
244        return Timestamp.from_pytime(value)
245
246    def to_pod(self,
247            context: typing.Union[edq.util.serial.SerializationContext, None] = None,
248            ) -> int:
249        return self
250
251    @classmethod
252    def from_pod(cls: typing.Type['Timestamp'],
253            data: typing.Any,
254            context: typing.Union[edq.util.serial.SerializationContext, None] = None,
255            ) -> 'Timestamp':
256        return cls.guess(data)
257
258def get_local_timezone() -> datetime.tzinfo:
259    """ Get the local (system) timezone or raise an exception. """
260
261    if (_testing_timezone is not None):
262        return _testing_timezone
263
264    local_timezone = datetime.datetime.now().astimezone().tzinfo
265    if ((local_timezone is None) or (not isinstance(local_timezone, datetime.tzinfo))):
266        raise ValueError("Could not discover local timezone.")
267
268    return local_timezone