edq.util.time

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