summaryrefslogtreecommitdiff
path: root/datetime_utils.py
blob: 795b427c31b4e57a4551aab26badd1f81b68c9a2 (plain)
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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
#!/usr/bin/env python3

"""Utilities related to dates and times and datetimes."""

import datetime
import enum
import logging
import re
from typing import Any, NewType, Tuple

import holidays  # type: ignore
import pytz

import constants

logger = logging.getLogger(__name__)


def replace_timezone(dt: datetime.datetime,
                     tz: datetime.tzinfo) -> datetime.datetime:
    return dt.replace(tzinfo=None).astimezone(tz=tz)


def now() -> datetime.datetime:
    return datetime.datetime.now()


def now_pst() -> datetime.datetime:
    return replace_timezone(now(), pytz.timezone("US/Pacific"))


def date_to_datetime(date: datetime.date) -> datetime.datetime:
    return datetime.datetime(
        date.year,
        date.month,
        date.day,
        0, 0, 0, 0
    )


def date_and_time_to_datetime(date: datetime.date,
                              time: datetime.time) -> datetime.datetime:
    return datetime.datetime(
        date.year,
        date.month,
        date.day,
        time.hour,
        time.minute,
        time.second,
        time.millisecond
    )


def datetime_to_date(date: datetime.datetime) -> datetime.date:
    return datetime.date(
        date.year,
        date.month,
        date.day
    )


# An enum to represent units with which we can compute deltas.
class TimeUnit(enum.Enum):
    MONDAYS = 0
    TUESDAYS = 1
    WEDNESDAYS = 2
    THURSDAYS = 3
    FRIDAYS = 4
    SATURDAYS = 5
    SUNDAYS = 6
    SECONDS = 7
    MINUTES = 8
    HOURS = 9
    DAYS = 10
    WORKDAYS = 11
    WEEKS = 12
    MONTHS = 13
    YEARS = 14

    @classmethod
    def is_valid(cls, value: Any):
        if type(value) is int:
            print("int")
            return value in cls._value2member_map_
        elif type(value) is TimeUnit:
            print("TimeUnit")
            return value.value in cls._value2member_map_
        elif type(value) is str:
            print("str")
            return value in cls._member_names_
        else:
            print(type(value))
            return False


def n_timeunits_from_base(
    count: int,
    unit: TimeUnit,
    base: datetime.datetime
) -> datetime.datetime:
    assert TimeUnit.is_valid(unit)
    if count == 0:
        return base

    # N days from base
    if unit == TimeUnit.DAYS:
        timedelta = datetime.timedelta(days=count)
        return base + timedelta

    # N workdays from base
    elif unit == TimeUnit.WORKDAYS:
        if count < 0:
            count = abs(count)
            timedelta = datetime.timedelta(days=-1)
        else:
            timedelta = datetime.timedelta(days=1)
        skips = holidays.US(years=base.year).keys()
        while count > 0:
            old_year = base.year
            base += timedelta
            if base.year != old_year:
                skips = holidays.US(years=base.year).keys()
            if (
                    base.weekday() < 5 and
                    datetime.date(base.year,
                                  base.month,
                                  base.day) not in skips
            ):
                count -= 1
        return base

    # N weeks from base
    elif unit == TimeUnit.WEEKS:
        timedelta = datetime.timedelta(weeks=count)
        base = base + timedelta
        return base

    # N months from base
    elif unit == TimeUnit.MONTHS:
        month_term = count % 12
        year_term = count // 12
        new_month = base.month + month_term
        if new_month > 12:
            new_month %= 12
            year_term += 1
        new_year = base.year + year_term
        return datetime.datetime(
            new_year,
            new_month,
            base.day,
            base.hour,
            base.minute,
            base.second,
            base.microsecond,
        )

    # N years from base
    elif unit == TimeUnit.YEARS:
        new_year = base.year + count
        return datetime.datetime(
            new_year,
            base.month,
            base.day,
            base.hour,
            base.minute,
            base.second,
            base.microsecond,
        )

    # N weekdays from base (e.g. 4 wednesdays from today)
    direction = 1 if count > 0 else -1
    count = abs(count)
    timedelta = datetime.timedelta(days=direction)
    start = base
    while True:
        dow = base.weekday()
        if dow == unit and start != base:
            count -= 1
            if count == 0:
                return base
        base = base + timedelta


def get_format_string(
        *,
        date_time_separator=" ",
        include_timezone=True,
        include_dayname=False,
        use_month_abbrevs=False,
        include_seconds=True,
        include_fractional=False,
        twelve_hour=True,
) -> str:
    fstring = ""
    if include_dayname:
        fstring += "%a/"

    if use_month_abbrevs:
        fstring = f"%Y/%b/%d{date_time_separator}"
    else:
        fstring = f"%Y/%m/%d{date_time_separator}"
    if twelve_hour:
        fstring += "%I:%M"
        if include_seconds:
            fstring += ":%S"
        fstring += "%p"
    else:
        fstring += "%H:%M"
        if include_seconds:
            fstring += ":%S"
    if include_fractional:
        fstring += ".%f"
    if include_timezone:
        fstring += "%z"
    return fstring


def datetime_to_string(
    dt: datetime.datetime,
    *,
    date_time_separator=" ",
    include_timezone=True,
    include_dayname=False,
    use_month_abbrevs=False,
    include_seconds=True,
    include_fractional=False,
    twelve_hour=True,
) -> str:
    """A nice way to convert a datetime into a string."""
    fstring = get_format_string(
        date_time_separator=date_time_separator,
        include_timezone=include_timezone,
        include_dayname=include_dayname,
        include_seconds=include_seconds,
        include_fractional=include_fractional,
        twelve_hour=twelve_hour)
    return dt.strftime(fstring).strip()


def string_to_datetime(
        txt: str,
        *,
        date_time_separator=" ",
        include_timezone=True,
        include_dayname=False,
        use_month_abbrevs=False,
        include_seconds=True,
        include_fractional=False,
        twelve_hour=True,
) -> Tuple[datetime.datetime, str]:
    """A nice way to convert a string into a datetime.  Also consider
    dateparse.dateparse_utils for a full parser.
    """
    fstring = get_format_string(
        date_time_separator=date_time_separator,
        include_timezone=include_timezone,
        include_dayname=include_dayname,
        include_seconds=include_seconds,
        include_fractional=include_fractional,
        twelve_hour=twelve_hour)
    return (
        datetime.datetime.strptime(txt, fstring),
        fstring
    )


def timestamp() -> str:
    """Return a timestamp for now in Pacific timezone."""
    ts = datetime.datetime.now(tz=pytz.timezone("US/Pacific"))
    return datetime_to_string(ts, include_timezone=True)


def time_to_string(
    dt: datetime.datetime,
    *,
    include_seconds=True,
    include_fractional=False,
    include_timezone=False,
    twelve_hour=True,
) -> str:
    """A nice way to convert a datetime into a time (only) string."""
    fstring = ""
    if twelve_hour:
        fstring += "%l:%M"
        if include_seconds:
            fstring += ":%S"
        fstring += "%p"
    else:
        fstring += "%H:%M"
        if include_seconds:
            fstring += ":%S"
    if include_fractional:
        fstring += ".%f"
    if include_timezone:
        fstring += "%z"
    return dt.strftime(fstring).strip()


def seconds_to_timedelta(seconds: int) -> datetime.timedelta:
    """Convert a delta in seconds into a timedelta."""
    return datetime.timedelta(seconds=seconds)


MinuteOfDay = NewType("MinuteOfDay", int)


def minute_number(hour: int, minute: int) -> MinuteOfDay:
    """Convert hour:minute into minute number from start of day."""
    return MinuteOfDay(hour * 60 + minute)


def datetime_to_minute_number(dt: datetime.datetime) -> MinuteOfDay:
    """Convert a datetime into a minute number (of the day)"""
    return minute_number(dt.hour, dt.minute)


def minute_number_to_time_string(minute_num: MinuteOfDay) -> str:
    """Convert minute number from start of day into hour:minute am/pm
    string.
    """
    hour = minute_num // 60
    minute = minute_num % 60
    ampm = "a"
    if hour > 12:
        hour -= 12
        ampm = "p"
    if hour == 12:
        ampm = "p"
    if hour == 0:
        hour = 12
    return f"{hour:2}:{minute:02}{ampm}"


def parse_duration(duration: str) -> int:
    """Parse a duration in string form."""
    seconds = 0
    m = re.search(r'(\d+) *d[ays]*', duration)
    if m is not None:
        seconds += int(m.group(1)) * 60 * 60 * 24
    m = re.search(r'(\d+) *h[ours]*', duration)
    if m is not None:
        seconds += int(m.group(1)) * 60 * 60
    m = re.search(r'(\d+) *m[inutes]*', duration)
    if m is not None:
        seconds += int(m.group(1)) * 60
    m = re.search(r'(\d+) *s[econds]*', duration)
    if m is not None:
        seconds += int(m.group(1))
    return seconds


def describe_duration(age: int) -> str:
    """Describe a duration."""
    days = divmod(age, constants.SECONDS_PER_DAY)
    hours = divmod(days[1], constants.SECONDS_PER_HOUR)
    minutes = divmod(hours[1], constants.SECONDS_PER_MINUTE)

    descr = ""
    if days[0] > 1:
        descr = f"{int(days[0])} days, "
    elif days[0] == 1:
        descr = "1 day, "
    if hours[0] > 1:
        descr = descr + f"{int(hours[0])} hours, "
    elif hours[0] == 1:
        descr = descr + "1 hour, "
    if len(descr) > 0:
        descr = descr + "and "
    if minutes[0] == 1:
        descr = descr + "1 minute"
    else:
        descr = descr + f"{int(minutes[0])} minutes"
    return descr


def describe_duration_briefly(age: int) -> str:
    """Describe a duration briefly."""
    days = divmod(age, constants.SECONDS_PER_DAY)
    hours = divmod(days[1], constants.SECONDS_PER_HOUR)
    minutes = divmod(hours[1], constants.SECONDS_PER_MINUTE)

    descr = ""
    if days[0] > 0:
        descr = f"{int(days[0])}d "
    if hours[0] > 0:
        descr = descr + f"{int(hours[0])}h "
    if minutes[0] > 0 or len(descr) == 0:
        descr = descr + f"{int(minutes[0])}m"
    return descr.strip()