Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dateutil/tz/_tzfile.py: 15%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# -*- coding: utf-8 -*-
2import bisect
3import calendar
4import functools
5import io
6import os
7import re
8import stat
9import struct
10import sys
11from datetime import MAXYEAR, MINYEAR, datetime, timedelta
13import six
15from ._common import _tzinfo, enfold, tzname_in_python2
17EPOCH = datetime(1970, 1, 1)
18EPOCHORDINAL = EPOCH.toordinal()
20# TODO: Move this to _common.py
21try:
22 # Python 3.7 feature
23 from contextlib import nullcontext as _nullcontext
24except ImportError:
25 import contextlib
27 @contextlib.contextmanager
28 def _nullcontext(context):
29 yield context
32# It is relatively expensive to construct new timedelta objects, and in most
33# cases we're looking at the same deltas, like integer numbers of hours, etc.
34# To improve speed and memory use, we'll keep a dictionary with references
35# to the ones we've already used so far.
36#
37# Loading every time zone in the 2020a version of the time zone database
38# requires 447 timedeltas, which requires approximately the amount of space
39# that ZoneInfo("America/New_York") with 236 transitions takes up, so we will
40# set the cache size to 512 so that in the common case we always get cache
41# hits, but specifically crafted ZoneInfo objects don't leak arbitrary amounts
42# of memory.
43_TIMEDELTA_CACHE_SIZE = 512
44if six.PY2:
46 def _cache_load_timedelta(f):
47 from collections import OrderedDict
49 from six.moves import _thread
51 CACHE_LOCK = _thread.allocate_lock()
52 TIMEDELTA_CACHE = OrderedDict()
54 @functools.wraps(f)
55 def cached_load_timedelta(seconds):
56 # OrderedDict is implemented in Python on Python 2, so every
57 # access has to hold the lock to keep its linked list intact.
58 with CACHE_LOCK:
59 rv = TIMEDELTA_CACHE.pop(seconds, None)
60 if rv is None:
61 rv = f(seconds)
63 TIMEDELTA_CACHE[seconds] = rv
65 if len(TIMEDELTA_CACHE) > _TIMEDELTA_CACHE_SIZE:
66 TIMEDELTA_CACHE.popitem(last=False)
68 return rv
70 return cached_load_timedelta
72else:
73 _cache_load_timedelta = functools.lru_cache(maxsize=512)
76@_cache_load_timedelta
77def _load_timedelta(seconds):
78 return timedelta(seconds=seconds)
81def _unpickle(cls, key, filename, data):
82 obj = cls.__new__(cls)
83 obj._key = key
84 obj._filename = filename
86 obj._load_from_data(data)
88 return obj
91class tzfile(_tzinfo):
92 """
93 This is a ``tzinfo`` subclass that allows one to use the ``tzfile(5)``
94 format timezone files to extract current and historical zone information.
96 :param fileobj:
97 This can be an opened file stream or a file name that the time zone
98 information can be read from.
100 :param filename:
101 This is an optional parameter specifying the source of the time zone
102 information in the event that ``fileobj`` is a file object. If omitted
103 and ``fileobj`` is a file stream, this parameter will be set either to
104 ``fileobj``'s ``name`` attribute or to ``repr(fileobj)``.
106 This class is intended to be mostly equivalent to the standard library
107 :py:class:`zoneinfo.ZoneInfo` class. The primary differences are:
109 * All caching in ``tzfile`` is handled by :func:`tz.gettz` level (you
110 should generally not attempt to construct a ``tzfile`` directly).
111 * ``tzfile`` objects are pickled by value rather than by reference to
112 the IANA key.
113 * ``tzfile`` has equality-by-value semantics, whereas
114 :py:class:`~zoneinfo.ZoneInfo` has equality-by-identity semantics.
116 See `Sources for Time Zone and Daylight Saving Time Data
117 <https://data.iana.org/time-zones/tz-link.html>`_ for more information.
118 Time zone files can be compiled from the `IANA Time Zone database files
119 <https://www.iana.org/time-zones>`_ with the `zic time zone compiler
120 <https://www.freebsd.org/cgi/man.cgi?query=zic&sektion=8>`_
122 .. note::
124 Only construct a ``tzfile`` directly if you have a specific timezone
125 file on disk that you want to read into a Python ``tzinfo`` object.
126 If you want to get a ``tzfile`` representing a specific IANA zone,
127 (e.g. ``'America/New_York'``), you should call
128 :func:`dateutil.tz.gettz` with the zone identifier.
130 .. versionchanged:: 3.0.0
131 ``tzfile`` is now a partial backport of :py:class:`zoneinfo.ZoneInfo`,
132 which brings with it support for Version 2 and 3 ``tzfile(5)`` binaries.
134 **Examples:**
136 Using the US Eastern time zone as an example, we can see that a ``tzfile``
137 provides time zone information for the standard Daylight Saving offsets:
139 .. testsetup:: tzfile
141 from dateutil.tz import gettz
142 from datetime import datetime
144 .. doctest:: tzfile
146 >>> NYC = gettz('America/New_York')
147 >>> NYC
148 tzfile('/usr/share/zoneinfo/America/New_York')
150 >>> print(datetime(2016, 1, 3, tzinfo=NYC)) # EST
151 2016-01-03 00:00:00-05:00
153 >>> print(datetime(2016, 7, 7, tzinfo=NYC)) # EDT
154 2016-07-07 00:00:00-04:00
157 The ``tzfile`` structure contains a fully history of the time zone,
158 so historical dates will also have the right offsets. For example, before
159 the adoption of standard time on November 18, 1883, New York used local
160 mean time:
162 .. doctest:: tzfile
164 >>> print(datetime(1883, 11, 18, tzinfo=NYC)) # LMT
165 1883-11-18 00:00:00-04:56:02
167 Python versions before 3.6 do not support sub-minute offsets, so there the
168 offset is rounded to ``-04:56``.
170 And during World War II, New York was on "Eastern War Time", which was a
171 state of permanent daylight saving time:
173 .. doctest:: tzfile
175 >>> print(datetime(1944, 2, 7, tzinfo=NYC)) # EWT
176 1944-02-07 00:00:00-04:00
178 """
180 def __init__(self, fileobj, filename=None, key=None):
181 super(tzfile, self).__init__()
183 file_opened_here = False
184 if isinstance(fileobj, six.string_types):
185 self._filename = fileobj
186 fileobj = open(fileobj, "rb")
187 file_opened_here = True
188 elif filename is not None:
189 self._filename = filename
190 elif hasattr(fileobj, "name"):
191 self._filename = fileobj.name
192 elif key is not None:
193 self._filename = key
194 else:
195 self._filename = repr(fileobj)
197 if fileobj is not None:
198 if not file_opened_here:
199 fileobj = _nullcontext(fileobj)
201 with fileobj as file_stream:
202 self._load_file(file_stream)
204 self._key = key
206 @property
207 def key(self):
208 """
209 Read-only attribute returning the ``key`` constructor argument.
211 For zones generated from :func:`gettz`, this will be the IANA key
212 passed to ``gettz`` — no normalization is done, so although in most
213 distributions of the time zone database "US/Eastern" is a link to
214 "America/New_York", ``tz.gettz("US/Eastern").key`` should always return
215 ``"US/Eastern"``.
217 For zones constructed via another mechanism without specifying ``key``,
218 this returns :obj:`None`.
219 """
220 return self._key
222 def is_ambiguous(self, dt):
223 """
224 Whether or not the "wall time" of a given datetime is ambiguous in this
225 zone.
227 :param dt:
228 A :py:class:`datetime.datetime`, naive or time zone aware.
231 :return:
232 Returns ``True`` if ambiguous, ``False`` otherwise.
234 .. versionadded:: 2.6.0
235 """
236 return self._is_ambiguous_or_imaginary(dt) == 1
238 def _is_ambiguous_or_imaginary(self, dt):
239 if dt is None or self._fixed_offset:
240 return 0
242 timestamp = self._get_local_timestamp(dt)
243 tti_0 = self._find_trans(dt, fold=0, timestamp=timestamp)
244 tti_1 = self._find_trans(dt, fold=1, timestamp=timestamp)
246 # Returns 1 if ambiguous, -1 if imaginary and 0 otherwise.
247 return int(tti_0.utcoff > tti_1.utcoff) - int(
248 tti_1.utcoff > tti_0.utcoff
249 )
251 def utcoffset(self, dt):
252 return self._find_trans(dt).utcoff
254 def dst(self, dt):
255 return self._find_trans(dt).dstoff
257 @tzname_in_python2
258 def tzname(self, dt):
259 return self._find_trans(dt).tzname
261 def fromutc(self, dt):
262 """
263 The ``tzfile`` implementation of :py:func:`datetime.tzinfo.fromutc`.
265 :param dt:
266 A :py:class:`datetime.datetime` object.
268 :raises TypeError:
269 Raised if ``dt`` is not a :py:class:`datetime.datetime` object.
271 :raises ValueError:
272 Raised if this is called with a ``dt`` which does not have this
273 ``tzinfo`` attached.
275 :return:
276 Returns a :py:class:`datetime.datetime` object representing the
277 wall time in ``self``'s time zone.
278 """
280 if not isinstance(dt, datetime):
281 raise TypeError("fromutc() requires a datetime argument")
282 if dt.tzinfo is not self:
283 raise ValueError("dt.tzinfo is not self")
285 timestamp = self._get_local_timestamp(dt)
286 num_trans = len(self._trans_utc)
288 if num_trans >= 1 and timestamp < self._trans_utc[0]:
289 tti = self._tti_before
290 fold = 0
291 elif (
292 num_trans == 0 or timestamp > self._trans_utc[-1]
293 ) and not isinstance(self._tz_after, _ttinfo):
294 tti, fold = self._tz_after.get_trans_info_fromutc(
295 timestamp, dt.year
296 )
298 # The TZ string only knows about its own transitions, so check
299 # whether we are in a fold created by the last explicit one.
300 if num_trans and not fold:
301 if num_trans > 1:
302 tti_prev = self._ttinfos[-2]
303 else:
304 tti_prev = self._tti_before
306 shift = tti_prev.utcoff - tti.utcoff
307 fold = shift.total_seconds() > timestamp - self._trans_utc[-1]
308 elif num_trans == 0:
309 tti = self._tz_after
310 fold = 0
311 else:
312 idx = bisect.bisect_right(self._trans_utc, timestamp)
314 if num_trans > 1 and timestamp >= self._trans_utc[1]:
315 tti_prev, tti = self._ttinfos[idx - 2 : idx]
316 elif timestamp > self._trans_utc[-1]:
317 # Only reachable with exactly one transition, so the
318 # offset before it is _tti_before, not _ttinfos[-1].
319 tti_prev = self._tti_before
320 tti = self._tz_after
321 else:
322 tti_prev = self._tti_before
323 tti = self._ttinfos[0]
325 # Detect fold
326 shift = tti_prev.utcoff - tti.utcoff
327 fold = shift.total_seconds() > timestamp - self._trans_utc[idx - 1]
328 dt += tti.utcoff
329 if fold:
330 return enfold(dt, fold=1)
331 else:
332 return dt
334 def _find_trans(self, dt, fold=None, timestamp=None):
335 if dt is None:
336 if self._fixed_offset:
337 return self._tz_after
338 else:
339 return _NO_TTINFO
341 if timestamp is None:
342 ts = self._get_local_timestamp(dt)
343 else:
344 ts = timestamp
346 if fold is None:
347 fold = getattr(dt, "fold", 0)
349 lt = self._trans_local[fold]
351 num_trans = len(lt)
353 if num_trans and ts < lt[0]:
354 return self._tti_before
355 elif not num_trans or ts > lt[-1]:
356 if isinstance(self._tz_after, _TZStr):
357 tti = self._tz_after.get_trans_info(ts, dt.year, fold)
359 # The TZ string only describes times after the last explicit
360 # transition. If its answer for this local time is a different
361 # offset from the last transition's, and corresponds to a time
362 # before that transition (e.g. when the TZ string's own DST
363 # period for that year ends after the transition), the offset
364 # from the last explicit transition applies instead.
365 last_tti = self._ttinfos[-1] if num_trans else None
366 if (
367 last_tti is not None
368 and tti.utcoff != last_tti.utcoff
369 and ts - tti.utcoff.total_seconds() < self._trans_utc[-1]
370 ):
371 return last_tti
373 return tti
374 else:
375 return self._tz_after
376 else:
377 # idx is the transition that occurs after this timestamp, so we
378 # subtract off 1 to get the current ttinfo
379 idx = bisect.bisect_right(lt, ts) - 1
380 assert idx >= 0
381 return self._ttinfos[idx]
383 def _get_local_timestamp(self, dt):
384 return (
385 (dt.toordinal() - EPOCHORDINAL) * 86400
386 + dt.hour * 3600
387 + dt.minute * 60
388 + dt.second
389 )
391 # The attributes set by _unpickle and _load_from_data
392 _UNPICKLE_ATTRS = frozenset(
393 [
394 "_filename",
395 "_fixed_offset",
396 "_isdst",
397 "_key",
398 "_trans_idx",
399 "_trans_local",
400 "_trans_utc",
401 "_tti_before",
402 "_ttinfo_list",
403 "_ttinfos",
404 "_tz_after",
405 "_tz_str",
406 "_utcoff_raw",
407 ]
408 )
410 def __reduce__(self):
411 return self.__reduce_ex__(None)
413 def __reduce_ex__(self, protocol):
414 utcoff = [
415 int(ttinfo.utcoff.total_seconds()) for ttinfo in self._ttinfo_list
416 ]
417 utcoff = getattr(self, "_utcoff_raw", utcoff)
418 abbr = [ttinfo.tzname for ttinfo in self._ttinfo_list]
420 # dst() is inferred per transition, so the transitions don't map back
421 # to the local time types, and these are kept as they were in the file
422 trans_idx = self._trans_idx
423 isdst = self._isdst
425 trans_utc = self._trans_utc
426 tz_str = self._tz_str
428 data = (trans_idx, trans_utc, utcoff, isdst, abbr, tz_str)
429 args = (self.__class__, self._key, self._filename, data)
431 # Preserve any other instance state, such as attributes added by a
432 # subclass, but not the attributes that _unpickle rebuilds from data
433 state = {
434 k: v for k, v in vars(self).items() if k not in self._UNPICKLE_ATTRS
435 }
436 if state:
437 return (_unpickle, args, state)
439 return (_unpickle, args)
441 def _load_file(self, fobj):
442 # Reading the file requires seeking past the parts we don't use and
443 # reading the footer a line at a time, so read anything that doesn't
444 # support that (e.g. a pipe) into memory first; TZif files are small.
445 if not _is_seekable(fobj) or not hasattr(fobj, "readline"):
446 fobj = io.BytesIO(fobj.read())
448 # Retrieve all the data as it exists in the zoneinfo file
449 data = load_data(fobj)
450 self._load_from_data(data)
452 def _load_from_data(self, data):
453 trans_idx, trans_utc, utcoff, isdst, abbr, tz_str = data
455 # Infer the DST offsets (needed for .dst()) from the data, for each
456 # transition and for each local time type
457 trans_dstoff = self._utcoff_to_dstoff(
458 trans_idx, trans_utc, utcoff, isdst
459 )
460 dstoff = [3600 if dst else 0 for dst in isdst]
461 for idx, trans_off in zip(trans_idx, trans_dstoff):
462 dstoff[idx] = trans_off
464 if sys.version_info < (3, 6):
465 # For python pre-3.6, round to full-minutes if that's not the case.
466 # Python's datetime doesn't accept sub-minute timezones. Check
467 # http://python.org/sf/1447945 or https://bugs.python.org/issue5288
468 # for some information.
469 utcoff_adjusted = [60 * ((s + 30) // 60) for s in utcoff]
470 dstoff = [60 * ((s + 30) // 60) for s in dstoff]
471 trans_dstoff = [60 * ((s + 30) // 60) for s in trans_dstoff]
472 self._utcoff_raw = utcoff
473 else:
474 utcoff_adjusted = utcoff
476 # Convert all the transition times (UTC) into "seconds since 1970-01-01 local time"
477 trans_local = self._ts_to_local(trans_idx, trans_utc, utcoff_adjusted)
479 # Construct `_ttinfo` objects for each transition in the file
480 _ttinfo_list = [
481 _ttinfo(
482 _load_timedelta(utcoffset), _load_timedelta(dstoffset), tzname
483 )
484 for utcoffset, dstoffset, tzname in zip(
485 utcoff_adjusted, dstoff, abbr
486 )
487 ]
489 self._ttinfo_list = _ttinfo_list
491 self._trans_idx = trans_idx
492 self._isdst = isdst
493 self._trans_utc = trans_utc
494 self._trans_local = trans_local
496 # Transitions to the same local time type usually have the same DST
497 # offset, so share the _ttinfo objects between them where possible.
498 ttinfos = {}
499 self._ttinfos = []
500 for idx, trans_off in zip(trans_idx, trans_dstoff):
501 tti = _ttinfo_list[idx]
502 if tti.dstoff.total_seconds() != trans_off:
503 key = (idx, trans_off)
504 if key not in ttinfos:
505 ttinfos[key] = _ttinfo(
506 tti.utcoff, _load_timedelta(trans_off), tti.tzname
507 )
508 tti = ttinfos[key]
509 self._ttinfos.append(tti)
511 # Find the first non-DST transition
512 for i in range(len(isdst)):
513 if not isdst[i]:
514 self._tti_before = _ttinfo_list[i]
515 break
516 else:
517 if self._ttinfos:
518 self._tti_before = self._ttinfos[0]
519 else:
520 self._tti_before = None
522 # Set the "fallback" time zone
523 self._tz_str = tz_str
524 if tz_str is not None and tz_str != b"":
525 self._tz_after = _parse_tz_str(tz_str.decode())
526 else:
527 if not self._ttinfos and not _ttinfo_list:
528 raise ValueError("No time zone information found.")
530 if self._ttinfos:
531 self._tz_after = self._ttinfos[-1]
532 else:
533 # With no transitions and no TZ string, RFC 9636 Section 3.2
534 # says that local time is specified by time type 0
535 self._tz_after = _ttinfo_list[0]
537 # Determine if this is a "fixed offset" zone, meaning that the output
538 # of the utcoffset, dst and tzname functions does not depend on the
539 # specific datetime passed.
540 #
541 # We make three simplifying assumptions here:
542 #
543 # 1. If _tz_after is not a _ttinfo, it has transitions that might
544 # actually occur (it is possible to construct TZ strings that
545 # specify STD and DST but no transitions ever occur, such as
546 # AAA0BBB,0/0,J365/25).
547 # 2. If _ttinfo_list contains more than one _ttinfo object, the objects
548 # represent different offsets.
549 # 3. _ttinfo_list contains no unused _ttinfos (in which case an
550 # otherwise fixed-offset zone with extra _ttinfos defined may
551 # appear to *not* be a fixed offset zone).
552 #
553 # Violations to these assumptions would be fairly exotic, and exotic
554 # zones should almost certainly not be used with datetime.time (the
555 # only thing that would be affected by this).
556 if len(_ttinfo_list) > 1 or not isinstance(self._tz_after, _ttinfo):
557 self._fixed_offset = False
558 elif not _ttinfo_list:
559 self._fixed_offset = True
560 else:
561 self._fixed_offset = _ttinfo_list[0] == self._tz_after
563 @staticmethod
564 def _utcoff_to_dstoff(trans_idx, trans_utc, utcoffsets, isdsts):
565 """Infer the DST offset (the value of dst()) for each transition.
567 TZif files give the total UTC offset of each local time type and
568 whether it is DST, but not how much of the offset is DST, so this has
569 to be worked out from the standard time on either side of each DST
570 period. That is done for each transition rather than for each local
571 time type, because zic reuses a local time type for periods with
572 different standard offsets (e.g. UTC+0 DST in the Azores was double
573 summer time in the 1940s and summer time later).
574 """
575 n = len(trans_idx)
577 # The DST offset implied by the last standard time before and the
578 # first standard time after each transition. Before the first
579 # transition, it's the first standard time type, as for _tti_before.
580 first_std_utcoff = next(
581 (utcoff for utcoff, dst in zip(utcoffsets, isdsts) if not dst),
582 None,
583 )
584 before = [None] * n
585 after = [None] * n
586 for indices, implied, std_utcoff in (
587 (range(n), before, first_std_utcoff),
588 (range(n - 1, -1, -1), after, None),
589 ):
590 for i in indices:
591 idx = trans_idx[i]
592 if not isdsts[idx]:
593 std_utcoff = utcoffsets[idx]
594 elif std_utcoff is not None:
595 implied[i] = utcoffsets[idx] - std_utcoff
597 # DST offsets where the standard time is the same on both sides
598 clear_cut = [
599 (trans_utc[i], before[i])
600 for i in range(n)
601 if before[i] and before[i] == after[i]
602 ]
604 dstoffs = [0] * n
605 for i in range(n):
606 if not isdsts[trans_idx[i]]:
607 continue
609 candidates = {c for c in (before[i], after[i]) if c}
610 if len(candidates) > 1:
611 # The standard offset changed during the DST period. A change
612 # from local mean time can give something like 0:34:39, which
613 # is not a plausible amount of DST.
614 plausible = {c for c in candidates if c % 300 == 0}
615 candidates = plausible or candidates
617 if len(candidates) > 1:
618 # Prefer the amount of DST the zone used around the same
619 # time, e.g. Tehran used 1:00 and Rarotonga 0:30, though the
620 # TZif data for Tehran in 1977 and for Rarotonga in 1978 look
621 # the same. This has to be local in time, because some zones
622 # used different amounts in different eras (e.g. Montevideo
623 # used 0:30 in the 1920s and 1:00 later).
624 nearby = {
625 dstoff
626 for ts, dstoff in clear_cut
627 if abs(ts - trans_utc[i]) <= _DST_CONSENSUS_WINDOW
628 }
629 consensus = candidates & nearby
630 if len(consensus) == 1:
631 candidates = consensus
633 if candidates:
634 # Otherwise, prefer positive DST, then the smaller amount. In
635 # the 1942 occupation zones, e.g. Europe/Kyiv, the standard
636 # time before gives -1:00 and the one after +1:00.
637 dstoffs[i] = min(candidates, key=lambda c: (c < 0, abs(c)))
638 else:
639 # Without any standard time to compare to, one hour is a much
640 # better guess than 0
641 dstoffs[i] = 3600
643 return dstoffs
645 @staticmethod
646 def _ts_to_local(trans_idx, trans_list_utc, utcoffsets):
647 """Generate number of seconds since 1970 *in the local time*.
649 This is necessary to easily find the transition times in local time"""
650 if not trans_list_utc:
651 return [[], []]
653 # Start with the timestamps and modify in-place
654 trans_list_wall = [list(trans_list_utc), list(trans_list_utc)]
656 if len(utcoffsets) > 1:
657 offset_0 = utcoffsets[0]
658 offset_1 = utcoffsets[trans_idx[0]]
659 if offset_1 > offset_0:
660 offset_1, offset_0 = offset_0, offset_1
661 else:
662 offset_0 = offset_1 = utcoffsets[0]
664 trans_list_wall[0][0] += offset_0
665 trans_list_wall[1][0] += offset_1
667 for i in range(1, len(trans_idx)):
668 offset_0 = utcoffsets[trans_idx[i - 1]]
669 offset_1 = utcoffsets[trans_idx[i]]
671 if offset_1 > offset_0:
672 offset_1, offset_0 = offset_0, offset_1
674 trans_list_wall[0][i] += offset_0
675 trans_list_wall[1][i] += offset_1
677 return trans_list_wall
679 def __eq__(self, other):
680 if not isinstance(other, tzfile):
681 return NotImplemented
683 return (
684 self._trans_utc == other._trans_utc
685 and self._trans_local == other._trans_local
686 and self._ttinfos == other._ttinfos
687 and self._tti_before == other._tti_before
688 and self._fallback_key() == other._fallback_key()
689 )
691 def _fallback_key(self):
692 # The behavior after the last transition is determined either by the
693 # TZ string or, in its absence, by a single _ttinfo; _TZStr objects
694 # do not define equality, so compare the string they came from.
695 if isinstance(self._tz_after, _ttinfo):
696 return self._tz_after
697 return self._tz_str
699 __hash__ = None
701 def __ne__(self, other):
702 return not (self == other)
704 def __repr__(self):
705 return "%s(%s)" % (self.__class__.__name__, repr(self._filename))
708# Before Python 3.16, pipes on Windows claim to be seekable, but seeking on
709# them silently does nothing. See https://github.com/python/cpython/issues/86768
710_SEEKABLE_PIPES_BUG = sys.platform == "win32" and sys.version_info < (3, 16)
713def _is_seekable(fobj):
714 if not hasattr(fobj, "seek"):
715 return False
717 seekable = getattr(fobj, "seekable", None)
718 # Python 2 file objects can seek but have no seekable() method
719 if seekable is not None and not seekable():
720 return False
722 if not _SEEKABLE_PIPES_BUG:
723 return True
725 # For anything backed by a file descriptor, only trust regular files.
726 try:
727 fd = fobj.fileno()
728 except (AttributeError, OSError, ValueError):
729 # io.UnsupportedOperation (e.g. from BytesIO) is both of the latter
730 return True
732 try:
733 return stat.S_ISREG(os.fstat(fd).st_mode)
734 except OSError:
735 return True
738def load_data(fobj):
739 try:
740 return _load_data(fobj)
741 except struct.error as e:
742 # struct.unpack fails when a read comes up short
743 six.raise_from(
744 ValueError("Invalid TZif file: unexpected end of file"), e
745 )
748def _load_data(fobj):
749 header = _TZifHeader.from_file(fobj)
751 if header.version == 1:
752 time_size = 4
753 time_type = "l"
754 else:
755 # Version 2+ has 64-bit integer transition times
756 time_size = 8
757 time_type = "q"
759 # Version 2+ also starts with a Version 1 header and data, which
760 # we need to skip now
761 skip_bytes = (
762 header.timecnt * 5 # Transition times and types
763 + header.typecnt * 6 # Local time type records
764 + header.charcnt # Time zone designations
765 + header.leapcnt * 8 # Leap second records
766 + header.isstdcnt # Standard/wall indicators
767 + header.isutcnt # UT/local indicators
768 )
770 fobj.seek(skip_bytes, 1)
772 # Now we need to read the second header, which is not the same
773 # as the first
774 header = _TZifHeader.from_file(fobj)
776 typecnt = header.typecnt
777 timecnt = header.timecnt
778 charcnt = header.charcnt
780 # The data portion starts with timecnt transitions and indices
781 if timecnt:
782 trans_list_utc = struct.unpack(
783 ">%s%s" % (timecnt, time_type), fobj.read(timecnt * time_size)
784 )
785 trans_idx = struct.unpack(">%sB" % timecnt, fobj.read(timecnt))
787 if max(trans_idx) >= typecnt:
788 raise ValueError(
789 "Invalid transition index found while reading TZif: %s"
790 % max(trans_idx)
791 )
792 else:
793 trans_list_utc = ()
794 trans_idx = ()
796 # Read the ttinfo struct, (utoff, isdst, abbrind)
797 if typecnt:
798 utcoff, isdst, abbrind = zip(
799 *(struct.unpack(">lbb", fobj.read(6)) for i in range(typecnt))
800 )
801 else:
802 utcoff = ()
803 isdst = ()
804 abbrind = ()
806 # Now read the abbreviations. They are null-terminated strings, indexed
807 # not by position in the array but by position in the unsplit
808 # abbreviation string. I suppose this makes more sense in C, which uses
809 # null to terminate the strings, but it's inconvenient here...
810 abbr_vals = {}
811 abbr_chars = fobj.read(charcnt)
813 def get_abbr(idx):
814 # Gets a string starting at idx and running until the next \x00
815 #
816 # We cannot pre-populate abbr_vals by splitting on \x00 because there
817 # are some zones that use subsets of longer abbreviations, like so:
818 #
819 # LMT\x00AHST\x00HDT\x00
820 #
821 # Where the idx to abbr mapping should be:
822 #
823 # {0: "LMT", 4: "AHST", 5: "HST", 9: "HDT"}
824 if idx not in abbr_vals:
825 span_end = abbr_chars.find(b"\x00", idx)
826 abbr_vals[idx] = abbr_chars[idx:span_end].decode()
828 return abbr_vals[idx]
830 abbr = tuple(get_abbr(idx) for idx in abbrind)
832 # The remainder of the file consists of leap seconds (currently unused) and
833 # the standard/wall and ut/local indicators, which are metadata we don't need.
834 # In version 2 files, we need to skip the unnecessary data to get at the TZ string,
835 # and in version 1 files we'll just hopefully consume the rest of the file.
837 # Each leap second record has size (time_size + 4)
838 skip_bytes = (
839 header.isutcnt + header.isstdcnt + header.leapcnt * (time_size + 4)
840 )
841 fobj.seek(skip_bytes, 1)
842 if header.version >= 2:
843 c = fobj.read(1) # Should be \n
844 if c != b"\n":
845 raise ValueError("Invalid TZif file: expected newline, got %r" % c)
847 line = fobj.readline()
848 if not line.endswith(b"\n"):
849 raise ValueError("Invalid TZif file: unexpected end of file")
850 tz_str = line.rstrip(b"\n")
851 else:
852 tz_str = None
854 return trans_idx, trans_list_utc, utcoff, isdst, abbr, tz_str
857# How close in time (in seconds) the DST periods that _utcoff_to_dstoff uses
858# to choose between two possible DST offsets have to be
859_DST_CONSENSUS_WINDOW = 10 * 365 * 86400
862class _TZifHeader(object):
863 __slots__ = [
864 "version",
865 "isutcnt",
866 "isstdcnt",
867 "leapcnt",
868 "timecnt",
869 "typecnt",
870 "charcnt",
871 ]
873 def __init__(self, *args):
874 assert len(self.__slots__) == len(args)
875 for attr, val in zip(self.__slots__, args):
876 setattr(self, attr, val)
878 @classmethod
879 def from_file(cls, stream):
880 # The header starts with a 4-byte "magic" value
881 if stream.read(4) != b"TZif":
882 raise ValueError("Invalid TZif file: magic not found")
884 _version = stream.read(1)
885 if _version == b"\x00":
886 version = 1
887 else:
888 version = int(_version)
889 stream.read(15)
891 args = (version,)
893 # Slots are defined in the order that the bytes are arranged
894 args = args + struct.unpack(">6l", stream.read(24))
896 return cls(*args)
899class _ttinfo(object):
900 __slots__ = ["utcoff", "dstoff", "tzname"]
902 def __init__(self, utcoff, dstoff, tzname):
903 self.utcoff = utcoff
904 self.dstoff = dstoff
905 self.tzname = tzname
907 def __eq__(self, other):
908 if not isinstance(other, _ttinfo):
909 return NotImplemented
911 return (
912 self.utcoff == other.utcoff
913 and self.dstoff == other.dstoff
914 and self.tzname == other.tzname
915 )
917 def __ne__(self, other):
918 # Python 2 does not derive __ne__ from __eq__
919 eq = self.__eq__(other)
920 if eq is NotImplemented:
921 return eq
922 return not eq
924 def __repr__(self): # pragma: nocover
925 return "%s(%s, %s, %s)" % (
926 self.__class__.__name__,
927 self.utcoff,
928 self.dstoff,
929 self.tzname,
930 )
933_NO_TTINFO = _ttinfo(None, None, None)
936def _unpack(t):
937 # Python 2-specific backport for x, *y = t
938 return t[0], t[1:]
941# re.ASCII does not exist on Python 2, where str patterns are ASCII-only
942# anyway (and unicode patterns are too unless re.UNICODE is passed).
943_RE_ASCII = getattr(re, "ASCII", 0)
946def _fullmatch(pattern, string, flags=0):
947 # Python 2-specific backport of re.fullmatch
948 return re.match(r"(?:%s)\Z" % pattern, string, flags)
951def _round_offset(seconds):
952 # Python < 3.6 does not support sub-minute offsets, see the comment in
953 # tzfile._load_from_data.
954 if sys.version_info < (3, 6):
955 return 60 * ((seconds + 30) // 60)
956 return seconds
959def _parse_tz_str(tz_str):
960 # The tz string has the format:
961 #
962 # std[offset[dst[offset],start[/time],end[/time]]]
963 #
964 # std and dst must be 3 or more characters long and must not contain
965 # a leading colon, embedded digits, commas, nor a plus or minus signs;
966 # The spaces between "std" and "offset" are only for display and are
967 # not actually present in the string.
968 #
969 # The format of the offset is ``[+|-]hh[:mm[:ss]]``
971 # TODO: When we drop Python 2, this should be offset_str, *start_end_str = ...
972 offset_str, start_end_str = _unpack(tz_str.split(",", 1))
974 parser_re = (
975 r"(?P<std>[a-zA-Z]+|<[a-zA-Z0-9+-]+>)"
976 r"(?:"
977 r"(?P<stdoff>[+-]?\d{1,3}(?::\d{2}(?::\d{2})?)?)"
978 r"(?:"
979 r"(?P<dst>[a-zA-Z]+|<[a-zA-Z0-9+-]+>)"
980 r"(?P<dstoff>[+-]?\d{1,3}(?::\d{2}(?::\d{2})?)?)?"
981 r")?" # dst
982 r")?" # stdoff
983 )
985 m = _fullmatch(parser_re, offset_str, _RE_ASCII)
987 if m is None:
988 raise ValueError("%s is not a valid TZ string" % tz_str)
990 std_abbr = m.group("std")
991 dst_abbr = m.group("dst")
992 dst_offset = None
994 std_abbr = std_abbr.strip("<>")
996 if dst_abbr:
997 dst_abbr = dst_abbr.strip("<>")
999 std_offset = m.group("stdoff")
1000 if std_offset:
1001 try:
1002 std_offset = _round_offset(_parse_tz_delta(std_offset))
1003 except ValueError as e:
1004 raise ValueError("Invalid STD offset in %s" % tz_str)
1005 else:
1006 # The STD offset is required
1007 raise ValueError("Invalid STD offset in %s" % tz_str)
1009 if dst_abbr is not None:
1010 dst_offset = m.group("dstoff")
1011 if dst_offset:
1012 try:
1013 dst_offset = _round_offset(_parse_tz_delta(dst_offset))
1014 except ValueError as e:
1015 raise ValueError("Invalid DST offset in %s" % tz_str)
1016 else:
1017 dst_offset = std_offset + 3600
1019 if not start_end_str:
1020 raise ValueError("Missing transition rules: %s" % tz_str)
1022 start_end_strs = start_end_str[0].split(",", 1)
1023 try:
1024 start, end = (_parse_dst_start_end(x) for x in start_end_strs)
1025 except ValueError as e:
1026 raise ValueError("Invalid TZ string: %s" % tz_str)
1028 return _TZStr(std_abbr, std_offset, dst_abbr, dst_offset, start, end)
1029 elif start_end_str:
1030 raise ValueError("Transition rule present without DST: %s" % tz_str)
1031 else:
1032 # This is a static ttinfo, don't return _TZStr
1033 return _ttinfo(
1034 _load_timedelta(std_offset), _load_timedelta(0), std_abbr
1035 )
1038def _parse_dst_start_end(dststr):
1039 date, time = _unpack(dststr.split("/", 1))
1040 date_type = date[:1]
1041 if date_type == "M":
1042 n_is_julian = False
1043 m = _fullmatch(r"M(\d{1,2})\.(\d)\.(\d)", date, _RE_ASCII)
1044 if m is None:
1045 raise ValueError("Invalid dst start/end date: %s" % dststr)
1046 date_offset = tuple(map(int, m.groups()))
1047 offset = _CalendarOffset(*date_offset)
1048 else:
1049 if date_type == "J":
1050 n_is_julian = True
1051 date = date[1:]
1052 else:
1053 n_is_julian = False
1055 if _fullmatch(r"\d{1,3}", date, _RE_ASCII) is None:
1056 raise ValueError("Invalid dst start/end date: %s" % dststr)
1057 doy = int(date)
1058 offset = _DayOffset(doy, n_is_julian)
1060 if time:
1061 offset.hour, offset.minute, offset.second = _parse_transition_time(
1062 time[0]
1063 )
1065 return offset
1068def _parse_transition_time(time_str):
1069 match = _fullmatch(
1070 r"(?P<sign>[+-])?(?P<h>\d{1,3})(:(?P<m>\d{2})(:(?P<s>\d{2}))?)?",
1071 time_str,
1072 _RE_ASCII,
1073 )
1074 if match is None:
1075 raise ValueError("Invalid time: %s" % time_str)
1077 h, m, s = (int(v or 0) for v in map(match.group, ("h", "m", "s")))
1079 if h > 167:
1080 raise ValueError("Hour must be in [0, 167]: %s" % time_str)
1082 if match.group("sign") == "-":
1083 h, m, s = -h, -m, -s
1085 return h, m, s
1088def _parse_tz_delta(tz_delta):
1089 match = _fullmatch(
1090 r"(?P<sign>[+-])?(?P<h>\d{1,3})(:(?P<m>\d{2})(:(?P<s>\d{2}))?)?",
1091 tz_delta,
1092 _RE_ASCII,
1093 )
1094 # Anything passed to this function should already have hit an equivalent
1095 # regular expression to find the section to parse.
1096 assert match is not None, tz_delta
1098 h, m, s = (int(v or 0) for v in map(match.group, ("h", "m", "s")))
1100 total = h * 3600 + m * 60 + s
1102 if h > 24:
1103 raise ValueError("Offset hours must be in [0, 24]: %s" % tz_delta)
1105 # Yes, +5 maps to an offset of -5h
1106 if match.group("sign") != "-":
1107 total *= -1
1109 return total
1112class _TZStr(object):
1113 __slots__ = (
1114 "std",
1115 "dst",
1116 "start",
1117 "end",
1118 "get_trans_info",
1119 "get_trans_info_fromutc",
1120 "dst_diff",
1121 )
1123 def __init__(
1124 self, std_abbr, std_offset, dst_abbr, dst_offset, start=None, end=None
1125 ):
1126 self.dst_diff = dst_offset - std_offset
1127 std_offset = _load_timedelta(std_offset)
1128 self.std = _ttinfo(
1129 utcoff=std_offset, dstoff=_load_timedelta(0), tzname=std_abbr
1130 )
1132 self.start = start
1133 self.end = end
1135 dst_offset = _load_timedelta(dst_offset)
1136 delta = _load_timedelta(self.dst_diff)
1137 self.dst = _ttinfo(utcoff=dst_offset, dstoff=delta, tzname=dst_abbr)
1139 # These are assertions because the constructor should only be called
1140 # by functions that would fail before passing start or end
1141 assert start is not None, "No transition start specified"
1142 assert end is not None, "No transition end specified"
1144 self.get_trans_info = self._get_trans_info
1145 self.get_trans_info_fromutc = self._get_trans_info_fromutc
1147 def transitions(self, year):
1148 start = self.start.year_to_epoch(year)
1149 end = self.end.year_to_epoch(year)
1150 return start, end
1152 def _near_new_year(self, ts, year):
1153 """Whether ts is close enough to the new year that the rules for the
1154 previous or next year might matter."""
1155 year_start = _post_epoch_days_before_year(year) * 86400
1156 year_end = _post_epoch_days_before_year(year + 1) * 86400
1157 return (
1158 ts < year_start + _NEW_YEAR_MARGIN
1159 or ts >= year_end - _NEW_YEAR_MARGIN
1160 )
1162 def _get_trans_info(self, ts, year, fold):
1163 """Get the information about the current transition - tti"""
1164 if self._near_new_year(ts, year):
1165 return self._get_trans_info_near_new_year(ts, fold)
1167 start, end = self.transitions(year)
1169 # With fold = 0, the period (denominated in local time) with the
1170 # smaller offset starts at the end of the gap and ends at the end of
1171 # the fold; with fold = 1, it runs from the start of the gap to the
1172 # beginning of the fold.
1173 #
1174 # So in order to determine the DST boundaries we need to know both
1175 # the fold and whether DST is positive or negative (rare), and it
1176 # turns out that this boils down to fold XOR is_positive.
1177 if fold == (self.dst_diff >= 0):
1178 end -= self.dst_diff
1179 else:
1180 start += self.dst_diff
1182 if start < end:
1183 isdst = start <= ts < end
1184 else:
1185 isdst = not (end <= ts < start)
1187 return self.dst if isdst else self.std
1189 def _get_trans_info_near_new_year(self, ts, fold):
1190 # Answer the question in terms of UTC, where the transitions from the
1191 # adjacent years are taken into account: a local time is valid for
1192 # an offset if the rules give that offset at the corresponding UTC
1193 # time.
1194 valid = []
1195 for tti in (self.std, self.dst):
1196 utc_ts = ts - tti.utcoff.total_seconds()
1197 try:
1198 utc_year = (EPOCH + timedelta(seconds=utc_ts)).year
1199 except OverflowError:
1200 # Before year 1 or after year 9999
1201 utc_year = MINYEAR if utc_ts < 0 else MAXYEAR
1202 actual, _ = self._get_trans_info_fromutc_near_new_year(
1203 utc_ts, utc_year
1204 )
1205 if actual is tti:
1206 valid.append(tti)
1208 if len(valid) == 1:
1209 return valid[0]
1211 smaller, larger = sorted((self.std, self.dst), key=lambda t: t.utcoff)
1212 if valid:
1213 # Ambiguous: fold=0 is the first occurrence, with the larger offset
1214 return larger if fold == 0 else smaller
1215 else:
1216 # Imaginary: fold=0 uses the offset from before the transition
1217 return smaller if fold == 0 else larger
1219 def _get_trans_info_fromutc_near_new_year(self, ts, year):
1220 # The rules are in terms of the local year, which near the new year
1221 # is not always the UTC year, and a rule can put a transition up to a
1222 # week into the previous or next year (e.g. "J365/25"), so look at the
1223 # transitions for the years on either side as well and use the last
1224 # one before ts.
1225 std_utcoff = self.std.utcoff.total_seconds()
1226 dst_utcoff = self.dst.utcoff.total_seconds()
1228 transitions = []
1229 for y in (year - 1, year, year + 1):
1230 start, end = self.transitions(y)
1231 transitions.append((start - std_utcoff, True))
1232 transitions.append((end - dst_utcoff, False))
1234 # When one DST period ends at the same instant as the next one starts,
1235 # as with "DST all year" rules like "0/0,J365/25", sorting puts the
1236 # end first, so these stay in DST without a transition in between.
1237 transitions.sort()
1239 isdst = not transitions[0][1]
1240 last_change = None
1241 i = 0
1242 while i < len(transitions) and transitions[i][0] <= ts:
1243 trans_ts = transitions[i][0]
1244 new_isdst = isdst
1245 while i < len(transitions) and transitions[i][0] == trans_ts:
1246 new_isdst = transitions[i][1]
1247 i += 1
1249 if new_isdst != isdst:
1250 isdst = new_isdst
1251 last_change = trans_ts
1253 if isdst:
1254 tti, tti_prev = self.dst, self.std
1255 else:
1256 tti, tti_prev = self.std, self.dst
1258 # A transition to a smaller offset creates a fold just after it.
1259 fold = False
1260 if last_change is not None:
1261 shift = (tti_prev.utcoff - tti.utcoff).total_seconds()
1262 fold = shift > ts - last_change
1264 return (tti, fold)
1266 def _get_trans_info_fromutc(self, ts, year):
1267 if self._near_new_year(ts, year):
1268 return self._get_trans_info_fromutc_near_new_year(ts, year)
1270 start, end = self.transitions(year)
1271 start -= self.std.utcoff.total_seconds()
1272 end -= self.dst.utcoff.total_seconds()
1274 if start < end:
1275 isdst = start <= ts < end
1276 else:
1277 isdst = not (end <= ts < start)
1279 # For positive DST, the ambiguous period is one dst_diff after the end
1280 # of DST; for negative DST, the ambiguous period is one dst_diff before
1281 # the start of DST.
1282 if self.dst_diff > 0:
1283 ambig_start = end
1284 ambig_end = end + self.dst_diff
1285 else:
1286 ambig_start = start
1287 ambig_end = start - self.dst_diff
1289 fold = ambig_start <= ts < ambig_end
1291 return (self.dst if isdst else self.std, fold)
1294# A rule can put a transition up to 167 hours (just under a week) away from
1295# the day it names, and local time is at most a day or so away from UTC, so
1296# only times within this distance of the new year can be affected by the
1297# rules for the previous or next year.
1298_NEW_YEAR_MARGIN = 10 * 86400
1301def _post_epoch_days_before_year(year):
1302 """Get the number of days between 1970-01-01 and YEAR-01-01"""
1303 y = year - 1
1304 return y * 365 + y // 4 - y // 100 + y // 400 + 1 - EPOCHORDINAL
1307class _DayOffset(object):
1308 __slots__ = ["d", "julian", "hour", "minute", "second"]
1310 def __init__(self, d, julian, hour=2, minute=0, second=0):
1311 min_day = 0 + julian # convert bool to int
1312 if not min_day <= d <= 365:
1313 raise ValueError("d must be in [%s, 365], not: %s" % (min_day, d))
1315 self.d = d
1316 self.julian = julian
1317 self.hour = hour
1318 self.minute = minute
1319 self.second = second
1321 def year_to_epoch(self, year):
1322 days_before_year = _post_epoch_days_before_year(year)
1324 # "n" counts days from 0 and includes February 29; "Jn" counts days
1325 # from 1 and never counts February 29, so in leap years, J60 (March 1)
1326 # and every day after it is one day later than its number suggests.
1327 d = self.d
1328 if self.julian:
1329 d -= 1
1330 if d >= 59 and calendar.isleap(year):
1331 d += 1
1333 epoch = (days_before_year + d) * 86400
1334 epoch += self.hour * 3600 + self.minute * 60 + self.second
1336 return epoch
1339class _CalendarOffset(object):
1340 __slots__ = ["m", "w", "d", "hour", "minute", "second"]
1342 _DAYS_BEFORE_MONTH = (
1343 -1,
1344 0,
1345 31,
1346 59,
1347 90,
1348 120,
1349 151,
1350 181,
1351 212,
1352 243,
1353 273,
1354 304,
1355 334,
1356 )
1358 def __init__(self, m, w, d, hour=2, minute=0, second=0):
1359 if not 1 <= m <= 12:
1360 raise ValueError("m must be in [1, 12]")
1362 if not 1 <= w <= 5:
1363 raise ValueError("w must be in [1, 5]")
1365 if not 0 <= d <= 6:
1366 raise ValueError("d must be in [0, 6]")
1368 self.m = m
1369 self.w = w
1370 self.d = d
1371 self.hour = hour
1372 self.minute = minute
1373 self.second = second
1375 @classmethod
1376 def _ymd2ord(cls, year, month, day):
1377 return (
1378 _post_epoch_days_before_year(year)
1379 + cls._DAYS_BEFORE_MONTH[month]
1380 + (month > 2 and calendar.isleap(year))
1381 + day
1382 - 1
1383 )
1385 # TODO: These are not actually epoch dates as they are expressed in local time
1386 def year_to_epoch(self, year):
1387 """Calculates the datetime of the occurrence from the year"""
1388 # We know year and month, we need to convert w, d into day of month
1389 #
1390 # Week 1 is the first week in which day `d` (where 0 = Sunday) appears.
1391 # Week 5 represents the last occurrence of day `d`, so we need to know
1392 # the range of the month.
1393 #
1394 # This doesn't use calendar.monthrange because the rules for years 0
1395 # and 10000 can affect times that datetime can represent, and
1396 # calendar.monthrange constructs a datetime.date. Instead, get the
1397 # weekday (0 = Monday) from the ordinal: 1970-01-01 was a Thursday.
1398 first_ordinal = self._ymd2ord(year, self.m, 1)
1399 first_day = (first_ordinal + 3) % 7
1400 if self.m == 12:
1401 next_ordinal = self._ymd2ord(year + 1, 1, 1)
1402 else:
1403 next_ordinal = self._ymd2ord(year, self.m + 1, 1)
1404 days_in_month = next_ordinal - first_ordinal
1406 # This equation seems magical, so I'll break it down:
1407 # 1. calendar says 0 = Monday, POSIX says 0 = Sunday
1408 # so we need first_day + 1 to get 1 = Monday -> 7 = Sunday,
1409 # which is still equivalent because this math is mod 7
1410 # 2. Get first day - desired day mod 7: -1 % 7 = 6, so we don't need
1411 # to do anything to adjust negative numbers.
1412 # 3. Add 1 because month days are a 1-based index.
1413 month_day = (self.d - (first_day + 1)) % 7 + 1
1415 # Now use a 0-based index version of `w` to calculate the w-th
1416 # occurrence of `d`
1417 month_day += (self.w - 1) * 7
1419 # month_day will only be > days_in_month if w was 5, and `w` means
1420 # "last occurrence of `d`", so now we just check if we over-shot the
1421 # end of the month and if so knock off 1 week.
1422 if month_day > days_in_month:
1423 month_day -= 7
1425 ordinal = self._ymd2ord(year, self.m, month_day)
1426 epoch = ordinal * 86400
1427 epoch += self.hour * 3600 + self.minute * 60 + self.second
1428 return epoch