Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dateutil/tz/_tzfile.py: 65%
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
1import bisect
2import calendar
3import contextlib
4import functools
5import io
6import os
7import re
8import stat
9import struct
10import sys
11from datetime import MAXYEAR, MINYEAR, datetime, timedelta
13from ._common import _tzinfo, enfold
15EPOCH = datetime(1970, 1, 1)
16EPOCHORDINAL = EPOCH.toordinal()
19# It is relatively expensive to construct new timedelta objects, and in most
20# cases we're looking at the same deltas, like integer numbers of hours, etc.
21# To improve speed and memory use, we'll keep a dictionary with references
22# to the ones we've already used so far.
23#
24# Loading every time zone in the 2020a version of the time zone database
25# requires 447 timedeltas, which requires approximately the amount of space
26# that ZoneInfo("America/New_York") with 236 transitions takes up, so we will
27# set the cache size to 512 so that in the common case we always get cache
28# hits, but specifically crafted ZoneInfo objects don't leak arbitrary amounts
29# of memory.
30_TIMEDELTA_CACHE_SIZE = 512
33@functools.lru_cache(maxsize=_TIMEDELTA_CACHE_SIZE)
34def _load_timedelta(seconds):
35 return timedelta(seconds=seconds)
38def _unpickle(cls, key, filename, data):
39 obj = cls.__new__(cls)
40 obj._key = key
41 obj._filename = filename
43 obj._load_from_data(data)
45 return obj
48class tzfile(_tzinfo):
49 """
50 This is a ``tzinfo`` subclass that allows one to use the ``tzfile(5)``
51 format timezone files to extract current and historical zone information.
53 :param fileobj:
54 This can be an opened file stream or a file name that the time zone
55 information can be read from.
57 :param filename:
58 This is an optional parameter specifying the source of the time zone
59 information in the event that ``fileobj`` is a file object. If omitted
60 and ``fileobj`` is a file stream, this parameter will be set either to
61 ``fileobj``'s ``name`` attribute or to ``repr(fileobj)``.
63 This class is intended to be mostly equivalent to the standard library
64 :py:class:`zoneinfo.ZoneInfo` class. The primary differences are:
66 * All caching in ``tzfile`` is handled by :func:`tz.gettz` level (you
67 should generally not attempt to construct a ``tzfile`` directly).
68 * ``tzfile`` objects are pickled by value rather than by reference to
69 the IANA key.
70 * ``tzfile`` has equality-by-value semantics, whereas
71 :py:class:`~zoneinfo.ZoneInfo` has equality-by-identity semantics.
73 See `Sources for Time Zone and Daylight Saving Time Data
74 <https://data.iana.org/time-zones/tz-link.html>`_ for more information.
75 Time zone files can be compiled from the `IANA Time Zone database files
76 <https://www.iana.org/time-zones>`_ with the `zic time zone compiler
77 <https://www.freebsd.org/cgi/man.cgi?query=zic&sektion=8>`_
79 .. note::
81 Only construct a ``tzfile`` directly if you have a specific timezone
82 file on disk that you want to read into a Python ``tzinfo`` object.
83 If you want to get a ``tzfile`` representing a specific IANA zone,
84 (e.g. ``'America/New_York'``), you should call
85 :func:`dateutil.tz.gettz` with the zone identifier.
87 .. versionchanged:: 3.0.0
88 ``tzfile`` is now a partial backport of :py:class:`zoneinfo.ZoneInfo`,
89 which brings with it support for Version 2 and 3 ``tzfile(5)`` binaries.
91 **Examples:**
93 Using the US Eastern time zone as an example, we can see that a ``tzfile``
94 provides time zone information for the standard Daylight Saving offsets:
96 .. testsetup:: tzfile
98 from dateutil.tz import gettz
99 from datetime import datetime
101 .. doctest:: tzfile
103 >>> NYC = gettz('America/New_York')
104 >>> NYC
105 tzfile('/usr/share/zoneinfo/America/New_York')
107 >>> print(datetime(2016, 1, 3, tzinfo=NYC)) # EST
108 2016-01-03 00:00:00-05:00
110 >>> print(datetime(2016, 7, 7, tzinfo=NYC)) # EDT
111 2016-07-07 00:00:00-04:00
114 The ``tzfile`` structure contains a fully history of the time zone,
115 so historical dates will also have the right offsets. For example, before
116 the adoption of standard time on November 18, 1883, New York used local
117 mean time:
119 .. doctest:: tzfile
121 >>> print(datetime(1883, 11, 18, tzinfo=NYC)) # LMT
122 1883-11-18 00:00:00-04:56:02
124 And during World War II, New York was on "Eastern War Time", which was a
125 state of permanent daylight saving time:
127 .. doctest:: tzfile
129 >>> print(datetime(1944, 2, 7, tzinfo=NYC)) # EWT
130 1944-02-07 00:00:00-04:00
132 """
134 def __init__(self, fileobj, filename=None, key=None):
135 super().__init__()
137 file_opened_here = False
138 if isinstance(fileobj, str):
139 self._filename = fileobj
140 fileobj = open(fileobj, "rb")
141 file_opened_here = True
142 elif filename is not None:
143 self._filename = filename
144 elif hasattr(fileobj, "name"):
145 self._filename = fileobj.name
146 elif key is not None:
147 self._filename = key
148 else:
149 self._filename = repr(fileobj)
151 if fileobj is not None:
152 if not file_opened_here:
153 fileobj = contextlib.nullcontext(fileobj)
155 with fileobj as file_stream:
156 self._load_file(file_stream)
158 self._key = key
160 @property
161 def key(self):
162 """
163 Read-only attribute returning the ``key`` constructor argument.
165 For zones generated from :func:`gettz`, this will be the IANA key
166 passed to ``gettz`` — no normalization is done, so although in most
167 distributions of the time zone database "US/Eastern" is a link to
168 "America/New_York", ``tz.gettz("US/Eastern").key`` should always return
169 ``"US/Eastern"``.
171 For zones constructed via another mechanism without specifying ``key``,
172 this returns :obj:`None`.
173 """
174 return self._key
176 def is_ambiguous(self, dt):
177 """
178 Whether or not the "wall time" of a given datetime is ambiguous in this
179 zone.
181 :param dt:
182 A :py:class:`datetime.datetime`, naive or time zone aware.
185 :return:
186 Returns ``True`` if ambiguous, ``False`` otherwise.
188 .. versionadded:: 2.6.0
189 """
190 return self._is_ambiguous_or_imaginary(dt) == 1
192 def _is_ambiguous_or_imaginary(self, dt):
193 if dt is None or self._fixed_offset:
194 return 0
196 timestamp = self._get_local_timestamp(dt)
197 tti_0 = self._find_trans(dt, fold=0, timestamp=timestamp)
198 tti_1 = self._find_trans(dt, fold=1, timestamp=timestamp)
200 # Returns 1 if ambiguous, -1 if imaginary and 0 otherwise.
201 return int(tti_0.utcoff > tti_1.utcoff) - int(
202 tti_1.utcoff > tti_0.utcoff
203 )
205 def utcoffset(self, dt):
206 return self._find_trans(dt).utcoff
208 def dst(self, dt):
209 return self._find_trans(dt).dstoff
211 def tzname(self, dt):
212 return self._find_trans(dt).tzname
214 def fromutc(self, dt):
215 """
216 The ``tzfile`` implementation of :py:func:`datetime.tzinfo.fromutc`.
218 :param dt:
219 A :py:class:`datetime.datetime` object.
221 :raises TypeError:
222 Raised if ``dt`` is not a :py:class:`datetime.datetime` object.
224 :raises ValueError:
225 Raised if this is called with a ``dt`` which does not have this
226 ``tzinfo`` attached.
228 :return:
229 Returns a :py:class:`datetime.datetime` object representing the
230 wall time in ``self``'s time zone.
231 """
233 if not isinstance(dt, datetime):
234 raise TypeError("fromutc() requires a datetime argument")
235 if dt.tzinfo is not self:
236 raise ValueError("dt.tzinfo is not self")
238 timestamp = self._get_local_timestamp(dt)
239 num_trans = len(self._trans_utc)
241 if num_trans >= 1 and timestamp < self._trans_utc[0]:
242 tti = self._tti_before
243 fold = 0
244 elif (
245 num_trans == 0 or timestamp > self._trans_utc[-1]
246 ) and not isinstance(self._tz_after, _ttinfo):
247 tti, fold = self._tz_after.get_trans_info_fromutc(
248 timestamp, dt.year
249 )
251 # The TZ string only knows about its own transitions, so check
252 # whether we are in a fold created by the last explicit one.
253 if num_trans and not fold:
254 if num_trans > 1:
255 tti_prev = self._ttinfos[-2]
256 else:
257 tti_prev = self._tti_before
259 shift = tti_prev.utcoff - tti.utcoff
260 fold = shift.total_seconds() > timestamp - self._trans_utc[-1]
261 elif num_trans == 0:
262 tti = self._tz_after
263 fold = 0
264 else:
265 idx = bisect.bisect_right(self._trans_utc, timestamp)
267 if num_trans > 1 and timestamp >= self._trans_utc[1]:
268 tti_prev, tti = self._ttinfos[idx - 2 : idx]
269 elif timestamp > self._trans_utc[-1]:
270 # Only reachable with exactly one transition, so the
271 # offset before it is _tti_before, not _ttinfos[-1].
272 tti_prev = self._tti_before
273 tti = self._tz_after
274 else:
275 tti_prev = self._tti_before
276 tti = self._ttinfos[0]
278 # Detect fold
279 shift = tti_prev.utcoff - tti.utcoff
280 fold = shift.total_seconds() > timestamp - self._trans_utc[idx - 1]
281 dt += tti.utcoff
282 if fold:
283 return enfold(dt, fold=1)
284 else:
285 return dt
287 def _find_trans(self, dt, fold=None, timestamp=None):
288 if dt is None:
289 if self._fixed_offset:
290 return self._tz_after
291 else:
292 return _NO_TTINFO
294 if timestamp is None:
295 ts = self._get_local_timestamp(dt)
296 else:
297 ts = timestamp
299 if fold is None:
300 fold = dt.fold
302 lt = self._trans_local[fold]
304 num_trans = len(lt)
306 if num_trans and ts < lt[0]:
307 return self._tti_before
308 elif not num_trans or ts > lt[-1]:
309 if isinstance(self._tz_after, _TZStr):
310 tti = self._tz_after.get_trans_info(ts, dt.year, fold)
312 # The TZ string only describes times after the last explicit
313 # transition. If its answer for this local time is a different
314 # offset from the last transition's, and corresponds to a time
315 # before that transition (e.g. when the TZ string's own DST
316 # period for that year ends after the transition), the offset
317 # from the last explicit transition applies instead.
318 last_tti = self._ttinfos[-1] if num_trans else None
319 if (
320 last_tti is not None
321 and tti.utcoff != last_tti.utcoff
322 and ts - tti.utcoff.total_seconds() < self._trans_utc[-1]
323 ):
324 return last_tti
326 return tti
327 else:
328 return self._tz_after
329 else:
330 # idx is the transition that occurs after this timestamp, so we
331 # subtract off 1 to get the current ttinfo
332 idx = bisect.bisect_right(lt, ts) - 1
333 assert idx >= 0
334 return self._ttinfos[idx]
336 def _get_local_timestamp(self, dt):
337 return (
338 (dt.toordinal() - EPOCHORDINAL) * 86400
339 + dt.hour * 3600
340 + dt.minute * 60
341 + dt.second
342 )
344 # The attributes set by _unpickle and _load_from_data
345 _UNPICKLE_ATTRS = frozenset(
346 [
347 "_filename",
348 "_fixed_offset",
349 "_isdst",
350 "_key",
351 "_trans_idx",
352 "_trans_local",
353 "_trans_utc",
354 "_tti_before",
355 "_ttinfo_list",
356 "_ttinfos",
357 "_tz_after",
358 "_tz_str",
359 ]
360 )
362 def __reduce__(self):
363 return self.__reduce_ex__(None)
365 def __reduce_ex__(self, protocol):
366 utcoff = [
367 int(ttinfo.utcoff.total_seconds()) for ttinfo in self._ttinfo_list
368 ]
369 abbr = [ttinfo.tzname for ttinfo in self._ttinfo_list]
371 # dst() is inferred per transition, so the transitions don't map back
372 # to the local time types, and these are kept as they were in the file
373 trans_idx = self._trans_idx
374 isdst = self._isdst
376 trans_utc = self._trans_utc
377 tz_str = self._tz_str
379 data = (trans_idx, trans_utc, utcoff, isdst, abbr, tz_str)
380 args = (self.__class__, self._key, self._filename, data)
382 # Preserve any other instance state, such as attributes added by a
383 # subclass, but not the attributes that _unpickle rebuilds from data
384 state = {
385 k: v for k, v in vars(self).items() if k not in self._UNPICKLE_ATTRS
386 }
387 if state:
388 return (_unpickle, args, state)
390 return (_unpickle, args)
392 def _load_file(self, fobj):
393 # Reading the file requires seeking past the parts we don't use and
394 # reading the footer a line at a time, so read anything that doesn't
395 # support that (e.g. a pipe) into memory first; TZif files are small.
396 if not _is_seekable(fobj) or not hasattr(fobj, "readline"):
397 fobj = io.BytesIO(fobj.read())
399 # Retrieve all the data as it exists in the zoneinfo file
400 data = load_data(fobj)
401 self._load_from_data(data)
403 def _load_from_data(self, data):
404 trans_idx, trans_utc, utcoff, isdst, abbr, tz_str = data
406 # Infer the DST offsets (needed for .dst()) from the data, for each
407 # transition and for each local time type
408 trans_dstoff = self._utcoff_to_dstoff(
409 trans_idx, trans_utc, utcoff, isdst
410 )
411 dstoff = [3600 if dst else 0 for dst in isdst]
412 for idx, trans_off in zip(trans_idx, trans_dstoff):
413 dstoff[idx] = trans_off
416 # Convert all the transition times (UTC) into "seconds since 1970-01-01 local time"
417 trans_local = self._ts_to_local(trans_idx, trans_utc, utcoff)
419 # Construct `_ttinfo` objects for each transition in the file
420 _ttinfo_list = [
421 _ttinfo(
422 _load_timedelta(utcoffset), _load_timedelta(dstoffset), tzname
423 )
424 for utcoffset, dstoffset, tzname in zip(utcoff, dstoff, abbr)
425 ]
427 self._ttinfo_list = _ttinfo_list
429 self._trans_idx = trans_idx
430 self._isdst = isdst
431 self._trans_utc = trans_utc
432 self._trans_local = trans_local
434 # Transitions to the same local time type usually have the same DST
435 # offset, so share the _ttinfo objects between them where possible.
436 ttinfos = {}
437 self._ttinfos = []
438 for idx, trans_off in zip(trans_idx, trans_dstoff):
439 tti = _ttinfo_list[idx]
440 if tti.dstoff.total_seconds() != trans_off:
441 key = (idx, trans_off)
442 if key not in ttinfos:
443 ttinfos[key] = _ttinfo(
444 tti.utcoff, _load_timedelta(trans_off), tti.tzname
445 )
446 tti = ttinfos[key]
447 self._ttinfos.append(tti)
449 # Find the first non-DST transition
450 for i in range(len(isdst)):
451 if not isdst[i]:
452 self._tti_before = _ttinfo_list[i]
453 break
454 else:
455 if self._ttinfos:
456 self._tti_before = self._ttinfos[0]
457 else:
458 self._tti_before = None
460 # Set the "fallback" time zone
461 self._tz_str = tz_str
462 if tz_str is not None and tz_str != b"":
463 self._tz_after = _parse_tz_str(tz_str.decode())
464 else:
465 if not self._ttinfos and not _ttinfo_list:
466 raise ValueError("No time zone information found.")
468 if self._ttinfos:
469 self._tz_after = self._ttinfos[-1]
470 else:
471 # With no transitions and no TZ string, RFC 9636 Section 3.2
472 # says that local time is specified by time type 0
473 self._tz_after = _ttinfo_list[0]
475 # Determine if this is a "fixed offset" zone, meaning that the output
476 # of the utcoffset, dst and tzname functions does not depend on the
477 # specific datetime passed.
478 #
479 # We make three simplifying assumptions here:
480 #
481 # 1. If _tz_after is not a _ttinfo, it has transitions that might
482 # actually occur (it is possible to construct TZ strings that
483 # specify STD and DST but no transitions ever occur, such as
484 # AAA0BBB,0/0,J365/25).
485 # 2. If _ttinfo_list contains more than one _ttinfo object, the objects
486 # represent different offsets.
487 # 3. _ttinfo_list contains no unused _ttinfos (in which case an
488 # otherwise fixed-offset zone with extra _ttinfos defined may
489 # appear to *not* be a fixed offset zone).
490 #
491 # Violations to these assumptions would be fairly exotic, and exotic
492 # zones should almost certainly not be used with datetime.time (the
493 # only thing that would be affected by this).
494 if len(_ttinfo_list) > 1 or not isinstance(self._tz_after, _ttinfo):
495 self._fixed_offset = False
496 elif not _ttinfo_list:
497 self._fixed_offset = True
498 else:
499 self._fixed_offset = _ttinfo_list[0] == self._tz_after
501 @staticmethod
502 def _utcoff_to_dstoff(trans_idx, trans_utc, utcoffsets, isdsts):
503 """Infer the DST offset (the value of dst()) for each transition.
505 TZif files give the total UTC offset of each local time type and
506 whether it is DST, but not how much of the offset is DST, so this has
507 to be worked out from the standard time on either side of each DST
508 period. That is done for each transition rather than for each local
509 time type, because zic reuses a local time type for periods with
510 different standard offsets (e.g. UTC+0 DST in the Azores was double
511 summer time in the 1940s and summer time later).
512 """
513 n = len(trans_idx)
515 # The DST offset implied by the last standard time before and the
516 # first standard time after each transition. Before the first
517 # transition, it's the first standard time type, as for _tti_before.
518 first_std_utcoff = next(
519 (utcoff for utcoff, dst in zip(utcoffsets, isdsts) if not dst),
520 None,
521 )
522 before = [None] * n
523 after = [None] * n
524 for indices, implied, std_utcoff in (
525 (range(n), before, first_std_utcoff),
526 (range(n - 1, -1, -1), after, None),
527 ):
528 for i in indices:
529 idx = trans_idx[i]
530 if not isdsts[idx]:
531 std_utcoff = utcoffsets[idx]
532 elif std_utcoff is not None:
533 implied[i] = utcoffsets[idx] - std_utcoff
535 # DST offsets where the standard time is the same on both sides
536 clear_cut = [
537 (trans_utc[i], before[i])
538 for i in range(n)
539 if before[i] and before[i] == after[i]
540 ]
542 dstoffs = [0] * n
543 for i in range(n):
544 if not isdsts[trans_idx[i]]:
545 continue
547 candidates = {c for c in (before[i], after[i]) if c}
548 if len(candidates) > 1:
549 # The standard offset changed during the DST period. A change
550 # from local mean time can give something like 0:34:39, which
551 # is not a plausible amount of DST.
552 plausible = {c for c in candidates if c % 300 == 0}
553 candidates = plausible or candidates
555 if len(candidates) > 1:
556 # Prefer the amount of DST the zone used around the same
557 # time, e.g. Tehran used 1:00 and Rarotonga 0:30, though the
558 # TZif data for Tehran in 1977 and for Rarotonga in 1978 look
559 # the same. This has to be local in time, because some zones
560 # used different amounts in different eras (e.g. Montevideo
561 # used 0:30 in the 1920s and 1:00 later).
562 nearby = {
563 dstoff
564 for ts, dstoff in clear_cut
565 if abs(ts - trans_utc[i]) <= _DST_CONSENSUS_WINDOW
566 }
567 consensus = candidates & nearby
568 if len(consensus) == 1:
569 candidates = consensus
571 if candidates:
572 # Otherwise, prefer positive DST, then the smaller amount. In
573 # the 1942 occupation zones, e.g. Europe/Kyiv, the standard
574 # time before gives -1:00 and the one after +1:00.
575 dstoffs[i] = min(candidates, key=lambda c: (c < 0, abs(c)))
576 else:
577 # Without any standard time to compare to, one hour is a much
578 # better guess than 0
579 dstoffs[i] = 3600
581 return dstoffs
583 @staticmethod
584 def _ts_to_local(trans_idx, trans_list_utc, utcoffsets):
585 """Generate number of seconds since 1970 *in the local time*.
587 This is necessary to easily find the transition times in local time"""
588 if not trans_list_utc:
589 return [[], []]
591 # Start with the timestamps and modify in-place
592 trans_list_wall = [list(trans_list_utc), list(trans_list_utc)]
594 if len(utcoffsets) > 1:
595 offset_0 = utcoffsets[0]
596 offset_1 = utcoffsets[trans_idx[0]]
597 if offset_1 > offset_0:
598 offset_1, offset_0 = offset_0, offset_1
599 else:
600 offset_0 = offset_1 = utcoffsets[0]
602 trans_list_wall[0][0] += offset_0
603 trans_list_wall[1][0] += offset_1
605 for i in range(1, len(trans_idx)):
606 offset_0 = utcoffsets[trans_idx[i - 1]]
607 offset_1 = utcoffsets[trans_idx[i]]
609 if offset_1 > offset_0:
610 offset_1, offset_0 = offset_0, offset_1
612 trans_list_wall[0][i] += offset_0
613 trans_list_wall[1][i] += offset_1
615 return trans_list_wall
617 def __eq__(self, other):
618 if not isinstance(other, tzfile):
619 return NotImplemented
621 return (
622 self._trans_utc == other._trans_utc
623 and self._trans_local == other._trans_local
624 and self._ttinfos == other._ttinfos
625 and self._tti_before == other._tti_before
626 and self._fallback_key() == other._fallback_key()
627 )
629 def _fallback_key(self):
630 # The behavior after the last transition is determined either by the
631 # TZ string or, in its absence, by a single _ttinfo; _TZStr objects
632 # do not define equality, so compare the string they came from.
633 if isinstance(self._tz_after, _ttinfo):
634 return self._tz_after
635 return self._tz_str
637 __hash__ = None
640 def __repr__(self):
641 return "%s(%s)" % (self.__class__.__name__, repr(self._filename))
644# Before Python 3.16, pipes on Windows claim to be seekable, but seeking on
645# them silently does nothing. See https://github.com/python/cpython/issues/86768
646_SEEKABLE_PIPES_BUG = sys.platform == "win32" and sys.version_info < (3, 16)
649def _is_seekable(fobj):
650 if not hasattr(fobj, "seek"):
651 return False
653 seekable = getattr(fobj, "seekable", None)
654 # File-like objects need not implement seekable().
655 if seekable is not None and not seekable():
656 return False
658 if not _SEEKABLE_PIPES_BUG:
659 return True
661 # For anything backed by a file descriptor, only trust regular files.
662 try:
663 fd = fobj.fileno()
664 except (AttributeError, OSError, ValueError):
665 # io.UnsupportedOperation (e.g. from BytesIO) is both of the latter
666 return True
668 try:
669 return stat.S_ISREG(os.fstat(fd).st_mode)
670 except OSError:
671 return True
674def load_data(fobj):
675 try:
676 return _load_data(fobj)
677 except struct.error as e:
678 # struct.unpack fails when a read comes up short
679 raise ValueError("Invalid TZif file: unexpected end of file") from e
682def _load_data(fobj):
683 header = _TZifHeader.from_file(fobj)
685 if header.version == 1:
686 time_size = 4
687 time_type = "l"
688 else:
689 # Version 2+ has 64-bit integer transition times
690 time_size = 8
691 time_type = "q"
693 # Version 2+ also starts with a Version 1 header and data, which
694 # we need to skip now
695 skip_bytes = (
696 header.timecnt * 5 # Transition times and types
697 + header.typecnt * 6 # Local time type records
698 + header.charcnt # Time zone designations
699 + header.leapcnt * 8 # Leap second records
700 + header.isstdcnt # Standard/wall indicators
701 + header.isutcnt # UT/local indicators
702 )
704 fobj.seek(skip_bytes, 1)
706 # Now we need to read the second header, which is not the same
707 # as the first
708 header = _TZifHeader.from_file(fobj)
710 typecnt = header.typecnt
711 timecnt = header.timecnt
712 charcnt = header.charcnt
714 # The data portion starts with timecnt transitions and indices
715 if timecnt:
716 trans_list_utc = struct.unpack(
717 ">%s%s" % (timecnt, time_type), fobj.read(timecnt * time_size)
718 )
719 trans_idx = struct.unpack(">%sB" % timecnt, fobj.read(timecnt))
721 if max(trans_idx) >= typecnt:
722 raise ValueError(
723 "Invalid transition index found while reading TZif: %s"
724 % max(trans_idx)
725 )
726 else:
727 trans_list_utc = ()
728 trans_idx = ()
730 # Read the ttinfo struct, (utoff, isdst, abbrind)
731 if typecnt:
732 utcoff, isdst, abbrind = zip(
733 *(struct.unpack(">lbb", fobj.read(6)) for i in range(typecnt))
734 )
735 else:
736 utcoff = ()
737 isdst = ()
738 abbrind = ()
740 # Now read the abbreviations. They are null-terminated strings, indexed
741 # not by position in the array but by position in the unsplit
742 # abbreviation string. I suppose this makes more sense in C, which uses
743 # null to terminate the strings, but it's inconvenient here...
744 abbr_vals = {}
745 abbr_chars = fobj.read(charcnt)
747 def get_abbr(idx):
748 # Gets a string starting at idx and running until the next \x00
749 #
750 # We cannot pre-populate abbr_vals by splitting on \x00 because there
751 # are some zones that use subsets of longer abbreviations, like so:
752 #
753 # LMT\x00AHST\x00HDT\x00
754 #
755 # Where the idx to abbr mapping should be:
756 #
757 # {0: "LMT", 4: "AHST", 5: "HST", 9: "HDT"}
758 if idx not in abbr_vals:
759 span_end = abbr_chars.find(b"\x00", idx)
760 abbr_vals[idx] = abbr_chars[idx:span_end].decode()
762 return abbr_vals[idx]
764 abbr = tuple(get_abbr(idx) for idx in abbrind)
766 # The remainder of the file consists of leap seconds (currently unused) and
767 # the standard/wall and ut/local indicators, which are metadata we don't need.
768 # In version 2 files, we need to skip the unnecessary data to get at the TZ string,
769 # and in version 1 files we'll just hopefully consume the rest of the file.
771 # Each leap second record has size (time_size + 4)
772 skip_bytes = (
773 header.isutcnt + header.isstdcnt + header.leapcnt * (time_size + 4)
774 )
775 fobj.seek(skip_bytes, 1)
776 if header.version >= 2:
777 c = fobj.read(1) # Should be \n
778 if c != b"\n":
779 raise ValueError("Invalid TZif file: expected newline, got %r" % c)
781 line = fobj.readline()
782 if not line.endswith(b"\n"):
783 raise ValueError("Invalid TZif file: unexpected end of file")
784 tz_str = line.rstrip(b"\n")
785 else:
786 tz_str = None
788 return trans_idx, trans_list_utc, utcoff, isdst, abbr, tz_str
791# How close in time (in seconds) the DST periods that _utcoff_to_dstoff uses
792# to choose between two possible DST offsets have to be
793_DST_CONSENSUS_WINDOW = 10 * 365 * 86400
796class _TZifHeader:
797 __slots__ = [
798 "version",
799 "isutcnt",
800 "isstdcnt",
801 "leapcnt",
802 "timecnt",
803 "typecnt",
804 "charcnt",
805 ]
807 def __init__(self, *args):
808 assert len(self.__slots__) == len(args)
809 for attr, val in zip(self.__slots__, args):
810 setattr(self, attr, val)
812 @classmethod
813 def from_file(cls, stream):
814 # The header starts with a 4-byte "magic" value
815 if stream.read(4) != b"TZif":
816 raise ValueError("Invalid TZif file: magic not found")
818 _version = stream.read(1)
819 if _version == b"\x00":
820 version = 1
821 else:
822 version = int(_version)
823 stream.read(15)
825 args = (version,)
827 # Slots are defined in the order that the bytes are arranged
828 args = args + struct.unpack(">6l", stream.read(24))
830 return cls(*args)
833class _ttinfo:
834 __slots__ = ["utcoff", "dstoff", "tzname"]
836 def __init__(self, utcoff, dstoff, tzname):
837 self.utcoff = utcoff
838 self.dstoff = dstoff
839 self.tzname = tzname
841 def __eq__(self, other):
842 if not isinstance(other, _ttinfo):
843 return NotImplemented
845 return (
846 self.utcoff == other.utcoff
847 and self.dstoff == other.dstoff
848 and self.tzname == other.tzname
849 )
852 def __repr__(self): # pragma: nocover
853 return "%s(%s, %s, %s)" % (
854 self.__class__.__name__,
855 self.utcoff,
856 self.dstoff,
857 self.tzname,
858 )
861_NO_TTINFO = _ttinfo(None, None, None)
864def _parse_tz_str(tz_str):
865 # The tz string has the format:
866 #
867 # std[offset[dst[offset],start[/time],end[/time]]]
868 #
869 # std and dst must be 3 or more characters long and must not contain
870 # a leading colon, embedded digits, commas, nor a plus or minus signs;
871 # The spaces between "std" and "offset" are only for display and are
872 # not actually present in the string.
873 #
874 # The format of the offset is ``[+|-]hh[:mm[:ss]]``
876 offset_str, *start_end_str = tz_str.split(",", 1)
878 parser_re = (
879 r"(?P<std>[a-zA-Z]+|<[a-zA-Z0-9+-]+>)"
880 r"(?:"
881 r"(?P<stdoff>[+-]?\d{1,3}(?::\d{2}(?::\d{2})?)?)"
882 r"(?:"
883 r"(?P<dst>[a-zA-Z]+|<[a-zA-Z0-9+-]+>)"
884 r"(?P<dstoff>[+-]?\d{1,3}(?::\d{2}(?::\d{2})?)?)?"
885 r")?" # dst
886 r")?" # stdoff
887 )
889 m = re.fullmatch(parser_re, offset_str, re.ASCII)
891 if m is None:
892 raise ValueError("%s is not a valid TZ string" % tz_str)
894 std_abbr = m.group("std")
895 dst_abbr = m.group("dst")
896 dst_offset = None
898 std_abbr = std_abbr.strip("<>")
900 if dst_abbr:
901 dst_abbr = dst_abbr.strip("<>")
903 std_offset = m.group("stdoff")
904 if std_offset:
905 try:
906 std_offset = _parse_tz_delta(std_offset)
907 except ValueError as e:
908 raise ValueError("Invalid STD offset in %s" % tz_str)
909 else:
910 # The STD offset is required
911 raise ValueError("Invalid STD offset in %s" % tz_str)
913 if dst_abbr is not None:
914 dst_offset = m.group("dstoff")
915 if dst_offset:
916 try:
917 dst_offset = _parse_tz_delta(dst_offset)
918 except ValueError as e:
919 raise ValueError("Invalid DST offset in %s" % tz_str)
920 else:
921 dst_offset = std_offset + 3600
923 if not start_end_str:
924 raise ValueError("Missing transition rules: %s" % tz_str)
926 start_end_strs = start_end_str[0].split(",", 1)
927 try:
928 start, end = (_parse_dst_start_end(x) for x in start_end_strs)
929 except ValueError as e:
930 raise ValueError("Invalid TZ string: %s" % tz_str)
932 return _TZStr(std_abbr, std_offset, dst_abbr, dst_offset, start, end)
933 elif start_end_str:
934 raise ValueError("Transition rule present without DST: %s" % tz_str)
935 else:
936 # This is a static ttinfo, don't return _TZStr
937 return _ttinfo(
938 _load_timedelta(std_offset), _load_timedelta(0), std_abbr
939 )
942def _parse_dst_start_end(dststr):
943 date, *time = dststr.split("/", 1)
944 date_type = date[:1]
945 if date_type == "M":
946 n_is_julian = False
947 m = re.fullmatch(r"M(\d{1,2})\.(\d)\.(\d)", date, re.ASCII)
948 if m is None:
949 raise ValueError("Invalid dst start/end date: %s" % dststr)
950 date_offset = tuple(map(int, m.groups()))
951 offset = _CalendarOffset(*date_offset)
952 else:
953 if date_type == "J":
954 n_is_julian = True
955 date = date[1:]
956 else:
957 n_is_julian = False
959 if re.fullmatch(r"\d{1,3}", date, re.ASCII) is None:
960 raise ValueError("Invalid dst start/end date: %s" % dststr)
961 doy = int(date)
962 offset = _DayOffset(doy, n_is_julian)
964 if time:
965 offset.hour, offset.minute, offset.second = _parse_transition_time(
966 time[0]
967 )
969 return offset
972def _parse_transition_time(time_str):
973 match = re.fullmatch(
974 r"(?P<sign>[+-])?(?P<h>\d{1,3})(:(?P<m>\d{2})(:(?P<s>\d{2}))?)?",
975 time_str,
976 re.ASCII,
977 )
978 if match is None:
979 raise ValueError("Invalid time: %s" % time_str)
981 h, m, s = (int(v or 0) for v in map(match.group, ("h", "m", "s")))
983 if h > 167:
984 raise ValueError("Hour must be in [0, 167]: %s" % time_str)
986 if match.group("sign") == "-":
987 h, m, s = -h, -m, -s
989 return h, m, s
992def _parse_tz_delta(tz_delta):
993 match = re.fullmatch(
994 r"(?P<sign>[+-])?(?P<h>\d{1,3})(:(?P<m>\d{2})(:(?P<s>\d{2}))?)?",
995 tz_delta,
996 re.ASCII,
997 )
998 # Anything passed to this function should already have hit an equivalent
999 # regular expression to find the section to parse.
1000 assert match is not None, tz_delta
1002 h, m, s = (int(v or 0) for v in map(match.group, ("h", "m", "s")))
1004 total = h * 3600 + m * 60 + s
1006 if h > 24:
1007 raise ValueError("Offset hours must be in [0, 24]: %s" % tz_delta)
1009 # Yes, +5 maps to an offset of -5h
1010 if match.group("sign") != "-":
1011 total *= -1
1013 return total
1016class _TZStr:
1017 __slots__ = (
1018 "std",
1019 "dst",
1020 "start",
1021 "end",
1022 "get_trans_info",
1023 "get_trans_info_fromutc",
1024 "dst_diff",
1025 )
1027 def __init__(
1028 self, std_abbr, std_offset, dst_abbr, dst_offset, start=None, end=None
1029 ):
1030 self.dst_diff = dst_offset - std_offset
1031 std_offset = _load_timedelta(std_offset)
1032 self.std = _ttinfo(
1033 utcoff=std_offset, dstoff=_load_timedelta(0), tzname=std_abbr
1034 )
1036 self.start = start
1037 self.end = end
1039 dst_offset = _load_timedelta(dst_offset)
1040 delta = _load_timedelta(self.dst_diff)
1041 self.dst = _ttinfo(utcoff=dst_offset, dstoff=delta, tzname=dst_abbr)
1043 # These are assertions because the constructor should only be called
1044 # by functions that would fail before passing start or end
1045 assert start is not None, "No transition start specified"
1046 assert end is not None, "No transition end specified"
1048 self.get_trans_info = self._get_trans_info
1049 self.get_trans_info_fromutc = self._get_trans_info_fromutc
1051 def transitions(self, year):
1052 start = self.start.year_to_epoch(year)
1053 end = self.end.year_to_epoch(year)
1054 return start, end
1056 def _near_new_year(self, ts, year):
1057 """Whether ts is close enough to the new year that the rules for the
1058 previous or next year might matter."""
1059 year_start = _post_epoch_days_before_year(year) * 86400
1060 year_end = _post_epoch_days_before_year(year + 1) * 86400
1061 return (
1062 ts < year_start + _NEW_YEAR_MARGIN
1063 or ts >= year_end - _NEW_YEAR_MARGIN
1064 )
1066 def _get_trans_info(self, ts, year, fold):
1067 """Get the information about the current transition - tti"""
1068 if self._near_new_year(ts, year):
1069 return self._get_trans_info_near_new_year(ts, fold)
1071 start, end = self.transitions(year)
1073 # With fold = 0, the period (denominated in local time) with the
1074 # smaller offset starts at the end of the gap and ends at the end of
1075 # the fold; with fold = 1, it runs from the start of the gap to the
1076 # beginning of the fold.
1077 #
1078 # So in order to determine the DST boundaries we need to know both
1079 # the fold and whether DST is positive or negative (rare), and it
1080 # turns out that this boils down to fold XOR is_positive.
1081 if fold == (self.dst_diff >= 0):
1082 end -= self.dst_diff
1083 else:
1084 start += self.dst_diff
1086 if start < end:
1087 isdst = start <= ts < end
1088 else:
1089 isdst = not (end <= ts < start)
1091 return self.dst if isdst else self.std
1093 def _get_trans_info_near_new_year(self, ts, fold):
1094 # Answer the question in terms of UTC, where the transitions from the
1095 # adjacent years are taken into account: a local time is valid for
1096 # an offset if the rules give that offset at the corresponding UTC
1097 # time.
1098 valid = []
1099 for tti in (self.std, self.dst):
1100 utc_ts = ts - tti.utcoff.total_seconds()
1101 try:
1102 utc_year = (EPOCH + timedelta(seconds=utc_ts)).year
1103 except OverflowError:
1104 # Before year 1 or after year 9999
1105 utc_year = MINYEAR if utc_ts < 0 else MAXYEAR
1106 actual, _ = self._get_trans_info_fromutc_near_new_year(
1107 utc_ts, utc_year
1108 )
1109 if actual is tti:
1110 valid.append(tti)
1112 if len(valid) == 1:
1113 return valid[0]
1115 smaller, larger = sorted((self.std, self.dst), key=lambda t: t.utcoff)
1116 if valid:
1117 # Ambiguous: fold=0 is the first occurrence, with the larger offset
1118 return larger if fold == 0 else smaller
1119 else:
1120 # Imaginary: fold=0 uses the offset from before the transition
1121 return smaller if fold == 0 else larger
1123 def _get_trans_info_fromutc_near_new_year(self, ts, year):
1124 # The rules are in terms of the local year, which near the new year
1125 # is not always the UTC year, and a rule can put a transition up to a
1126 # week into the previous or next year (e.g. "J365/25"), so look at the
1127 # transitions for the years on either side as well and use the last
1128 # one before ts.
1129 std_utcoff = self.std.utcoff.total_seconds()
1130 dst_utcoff = self.dst.utcoff.total_seconds()
1132 transitions = []
1133 for y in (year - 1, year, year + 1):
1134 start, end = self.transitions(y)
1135 transitions.append((start - std_utcoff, True))
1136 transitions.append((end - dst_utcoff, False))
1138 # When one DST period ends at the same instant as the next one starts,
1139 # as with "DST all year" rules like "0/0,J365/25", sorting puts the
1140 # end first, so these stay in DST without a transition in between.
1141 transitions.sort()
1143 isdst = not transitions[0][1]
1144 last_change = None
1145 i = 0
1146 while i < len(transitions) and transitions[i][0] <= ts:
1147 trans_ts = transitions[i][0]
1148 new_isdst = isdst
1149 while i < len(transitions) and transitions[i][0] == trans_ts:
1150 new_isdst = transitions[i][1]
1151 i += 1
1153 if new_isdst != isdst:
1154 isdst = new_isdst
1155 last_change = trans_ts
1157 if isdst:
1158 tti, tti_prev = self.dst, self.std
1159 else:
1160 tti, tti_prev = self.std, self.dst
1162 # A transition to a smaller offset creates a fold just after it.
1163 fold = False
1164 if last_change is not None:
1165 shift = (tti_prev.utcoff - tti.utcoff).total_seconds()
1166 fold = shift > ts - last_change
1168 return (tti, fold)
1170 def _get_trans_info_fromutc(self, ts, year):
1171 if self._near_new_year(ts, year):
1172 return self._get_trans_info_fromutc_near_new_year(ts, year)
1174 start, end = self.transitions(year)
1175 start -= self.std.utcoff.total_seconds()
1176 end -= self.dst.utcoff.total_seconds()
1178 if start < end:
1179 isdst = start <= ts < end
1180 else:
1181 isdst = not (end <= ts < start)
1183 # For positive DST, the ambiguous period is one dst_diff after the end
1184 # of DST; for negative DST, the ambiguous period is one dst_diff before
1185 # the start of DST.
1186 if self.dst_diff > 0:
1187 ambig_start = end
1188 ambig_end = end + self.dst_diff
1189 else:
1190 ambig_start = start
1191 ambig_end = start - self.dst_diff
1193 fold = ambig_start <= ts < ambig_end
1195 return (self.dst if isdst else self.std, fold)
1198# A rule can put a transition up to 167 hours (just under a week) away from
1199# the day it names, and local time is at most a day or so away from UTC, so
1200# only times within this distance of the new year can be affected by the
1201# rules for the previous or next year.
1202_NEW_YEAR_MARGIN = 10 * 86400
1205def _post_epoch_days_before_year(year):
1206 """Get the number of days between 1970-01-01 and YEAR-01-01"""
1207 y = year - 1
1208 return y * 365 + y // 4 - y // 100 + y // 400 + 1 - EPOCHORDINAL
1211class _DayOffset:
1212 __slots__ = ["d", "julian", "hour", "minute", "second"]
1214 def __init__(self, d, julian, hour=2, minute=0, second=0):
1215 min_day = 0 + julian # convert bool to int
1216 if not min_day <= d <= 365:
1217 raise ValueError("d must be in [%s, 365], not: %s" % (min_day, d))
1219 self.d = d
1220 self.julian = julian
1221 self.hour = hour
1222 self.minute = minute
1223 self.second = second
1225 def year_to_epoch(self, year):
1226 days_before_year = _post_epoch_days_before_year(year)
1228 # "n" counts days from 0 and includes February 29; "Jn" counts days
1229 # from 1 and never counts February 29, so in leap years, J60 (March 1)
1230 # and every day after it is one day later than its number suggests.
1231 d = self.d
1232 if self.julian:
1233 d -= 1
1234 if d >= 59 and calendar.isleap(year):
1235 d += 1
1237 epoch = (days_before_year + d) * 86400
1238 epoch += self.hour * 3600 + self.minute * 60 + self.second
1240 return epoch
1243class _CalendarOffset:
1244 __slots__ = ["m", "w", "d", "hour", "minute", "second"]
1246 _DAYS_BEFORE_MONTH = (
1247 -1,
1248 0,
1249 31,
1250 59,
1251 90,
1252 120,
1253 151,
1254 181,
1255 212,
1256 243,
1257 273,
1258 304,
1259 334,
1260 )
1262 def __init__(self, m, w, d, hour=2, minute=0, second=0):
1263 if not 1 <= m <= 12:
1264 raise ValueError("m must be in [1, 12]")
1266 if not 1 <= w <= 5:
1267 raise ValueError("w must be in [1, 5]")
1269 if not 0 <= d <= 6:
1270 raise ValueError("d must be in [0, 6]")
1272 self.m = m
1273 self.w = w
1274 self.d = d
1275 self.hour = hour
1276 self.minute = minute
1277 self.second = second
1279 @classmethod
1280 def _ymd2ord(cls, year, month, day):
1281 return (
1282 _post_epoch_days_before_year(year)
1283 + cls._DAYS_BEFORE_MONTH[month]
1284 + (month > 2 and calendar.isleap(year))
1285 + day
1286 - 1
1287 )
1289 # TODO: These are not actually epoch dates as they are expressed in local time
1290 def year_to_epoch(self, year):
1291 """Calculates the datetime of the occurrence from the year"""
1292 # We know year and month, we need to convert w, d into day of month
1293 #
1294 # Week 1 is the first week in which day `d` (where 0 = Sunday) appears.
1295 # Week 5 represents the last occurrence of day `d`, so we need to know
1296 # the range of the month.
1297 #
1298 # This doesn't use calendar.monthrange because the rules for years 0
1299 # and 10000 can affect times that datetime can represent, and
1300 # calendar.monthrange constructs a datetime.date. Instead, get the
1301 # weekday (0 = Monday) from the ordinal: 1970-01-01 was a Thursday.
1302 first_ordinal = self._ymd2ord(year, self.m, 1)
1303 first_day = (first_ordinal + 3) % 7
1304 if self.m == 12:
1305 next_ordinal = self._ymd2ord(year + 1, 1, 1)
1306 else:
1307 next_ordinal = self._ymd2ord(year, self.m + 1, 1)
1308 days_in_month = next_ordinal - first_ordinal
1310 # This equation seems magical, so I'll break it down:
1311 # 1. calendar says 0 = Monday, POSIX says 0 = Sunday
1312 # so we need first_day + 1 to get 1 = Monday -> 7 = Sunday,
1313 # which is still equivalent because this math is mod 7
1314 # 2. Get first day - desired day mod 7: -1 % 7 = 6, so we don't need
1315 # to do anything to adjust negative numbers.
1316 # 3. Add 1 because month days are a 1-based index.
1317 month_day = (self.d - (first_day + 1)) % 7 + 1
1319 # Now use a 0-based index version of `w` to calculate the w-th
1320 # occurrence of `d`
1321 month_day += (self.w - 1) * 7
1323 # month_day will only be > days_in_month if w was 5, and `w` means
1324 # "last occurrence of `d`", so now we just check if we over-shot the
1325 # end of the month and if so knock off 1 week.
1326 if month_day > days_in_month:
1327 month_day -= 7
1329 ordinal = self._ymd2ord(year, self.m, month_day)
1330 epoch = ordinal * 86400
1331 epoch += self.hour * 3600 + self.minute * 60 + self.second
1332 return epoch