diff options
Diffstat (limited to 'pendulum/tz')
| -rw-r--r-- | pendulum/tz/__init__.py | 60 | ||||
| -rw-r--r-- | pendulum/tz/data/__init__.py | 0 | ||||
| -rw-r--r-- | pendulum/tz/data/windows.py | 137 | ||||
| -rw-r--r-- | pendulum/tz/exceptions.py | 23 | ||||
| -rw-r--r-- | pendulum/tz/local_timezone.py | 257 | ||||
| -rw-r--r-- | pendulum/tz/timezone.py | 377 | ||||
| -rw-r--r-- | pendulum/tz/zoneinfo/__init__.py | 16 | ||||
| -rw-r--r-- | pendulum/tz/zoneinfo/exceptions.py | 18 | ||||
| -rw-r--r-- | pendulum/tz/zoneinfo/posix_timezone.py | 270 | ||||
| -rw-r--r-- | pendulum/tz/zoneinfo/reader.py | 224 | ||||
| -rw-r--r-- | pendulum/tz/zoneinfo/timezone.py | 128 | ||||
| -rw-r--r-- | pendulum/tz/zoneinfo/transition.py | 77 | ||||
| -rw-r--r-- | pendulum/tz/zoneinfo/transition_type.py | 35 |
13 files changed, 1622 insertions, 0 deletions
diff --git a/pendulum/tz/__init__.py b/pendulum/tz/__init__.py new file mode 100644 index 0000000..b085f37 --- /dev/null +++ b/pendulum/tz/__init__.py @@ -0,0 +1,60 @@ +from typing import Tuple
+from typing import Union
+
+import pytzdata
+
+from .local_timezone import get_local_timezone
+from .local_timezone import set_local_timezone
+from .local_timezone import test_local_timezone
+from .timezone import UTC
+from .timezone import FixedTimezone as _FixedTimezone
+from .timezone import Timezone as _Timezone
+
+
+PRE_TRANSITION = "pre"
+POST_TRANSITION = "post"
+TRANSITION_ERROR = "error"
+
+timezones = pytzdata.timezones # type: Tuple[str, ...]
+
+
+_tz_cache = {}
+
+
+def timezone(name, extended=True): # type: (Union[str, int], bool) -> _Timezone
+ """
+ Return a Timezone instance given its name.
+ """
+ if isinstance(name, int):
+ return fixed_timezone(name)
+
+ if name.lower() == "utc":
+ return UTC
+
+ if name in _tz_cache:
+ return _tz_cache[name]
+
+ tz = _Timezone(name, extended=extended)
+ _tz_cache[name] = tz
+
+ return tz
+
+
+def fixed_timezone(offset): # type: (int) -> _FixedTimezone
+ """
+ Return a Timezone instance given its offset in seconds.
+ """
+ if offset in _tz_cache:
+ return _tz_cache[offset] # type: ignore
+
+ tz = _FixedTimezone(offset)
+ _tz_cache[offset] = tz
+
+ return tz
+
+
+def local_timezone(): # type: () -> _Timezone
+ """
+ Return the local timezone.
+ """
+ return get_local_timezone()
diff --git a/pendulum/tz/data/__init__.py b/pendulum/tz/data/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/pendulum/tz/data/__init__.py diff --git a/pendulum/tz/data/windows.py b/pendulum/tz/data/windows.py new file mode 100644 index 0000000..7fb5b32 --- /dev/null +++ b/pendulum/tz/data/windows.py @@ -0,0 +1,137 @@ +windows_timezones = {
+ "AUS Central Standard Time": "Australia/Darwin",
+ "AUS Eastern Standard Time": "Australia/Sydney",
+ "Afghanistan Standard Time": "Asia/Kabul",
+ "Alaskan Standard Time": "America/Anchorage",
+ "Aleutian Standard Time": "America/Adak",
+ "Altai Standard Time": "Asia/Barnaul",
+ "Arab Standard Time": "Asia/Riyadh",
+ "Arabian Standard Time": "Asia/Dubai",
+ "Arabic Standard Time": "Asia/Baghdad",
+ "Argentina Standard Time": "America/Buenos_Aires",
+ "Astrakhan Standard Time": "Europe/Astrakhan",
+ "Atlantic Standard Time": "America/Halifax",
+ "Aus Central W. Standard Time": "Australia/Eucla",
+ "Azerbaijan Standard Time": "Asia/Baku",
+ "Azores Standard Time": "Atlantic/Azores",
+ "Bahia Standard Time": "America/Bahia",
+ "Bangladesh Standard Time": "Asia/Dhaka",
+ "Belarus Standard Time": "Europe/Minsk",
+ "Bougainville Standard Time": "Pacific/Bougainville",
+ "Canada Central Standard Time": "America/Regina",
+ "Cape Verde Standard Time": "Atlantic/Cape_Verde",
+ "Caucasus Standard Time": "Asia/Yerevan",
+ "Cen. Australia Standard Time": "Australia/Adelaide",
+ "Central America Standard Time": "America/Guatemala",
+ "Central Asia Standard Time": "Asia/Almaty",
+ "Central Brazilian Standard Time": "America/Cuiaba",
+ "Central Europe Standard Time": "Europe/Budapest",
+ "Central European Standard Time": "Europe/Warsaw",
+ "Central Pacific Standard Time": "Pacific/Guadalcanal",
+ "Central Standard Time": "America/Chicago",
+ "Central Standard Time (Mexico)": "America/Mexico_City",
+ "Chatham Islands Standard Time": "Pacific/Chatham",
+ "China Standard Time": "Asia/Shanghai",
+ "Cuba Standard Time": "America/Havana",
+ "Dateline Standard Time": "Etc/GMT+12",
+ "E. Africa Standard Time": "Africa/Nairobi",
+ "E. Australia Standard Time": "Australia/Brisbane",
+ "E. Europe Standard Time": "Europe/Chisinau",
+ "E. South America Standard Time": "America/Sao_Paulo",
+ "Easter Island Standard Time": "Pacific/Easter",
+ "Eastern Standard Time": "America/New_York",
+ "Eastern Standard Time (Mexico)": "America/Cancun",
+ "Egypt Standard Time": "Africa/Cairo",
+ "Ekaterinburg Standard Time": "Asia/Yekaterinburg",
+ "FLE Standard Time": "Europe/Kiev",
+ "Fiji Standard Time": "Pacific/Fiji",
+ "GMT Standard Time": "Europe/London",
+ "GTB Standard Time": "Europe/Bucharest",
+ "Georgian Standard Time": "Asia/Tbilisi",
+ "Greenland Standard Time": "America/Godthab",
+ "Greenwich Standard Time": "Atlantic/Reykjavik",
+ "Haiti Standard Time": "America/Port-au-Prince",
+ "Hawaiian Standard Time": "Pacific/Honolulu",
+ "India Standard Time": "Asia/Calcutta",
+ "Iran Standard Time": "Asia/Tehran",
+ "Israel Standard Time": "Asia/Jerusalem",
+ "Jordan Standard Time": "Asia/Amman",
+ "Kaliningrad Standard Time": "Europe/Kaliningrad",
+ "Korea Standard Time": "Asia/Seoul",
+ "Libya Standard Time": "Africa/Tripoli",
+ "Line Islands Standard Time": "Pacific/Kiritimati",
+ "Lord Howe Standard Time": "Australia/Lord_Howe",
+ "Magadan Standard Time": "Asia/Magadan",
+ "Magallanes Standard Time": "America/Punta_Arenas",
+ "Marquesas Standard Time": "Pacific/Marquesas",
+ "Mauritius Standard Time": "Indian/Mauritius",
+ "Middle East Standard Time": "Asia/Beirut",
+ "Montevideo Standard Time": "America/Montevideo",
+ "Morocco Standard Time": "Africa/Casablanca",
+ "Mountain Standard Time": "America/Denver",
+ "Mountain Standard Time (Mexico)": "America/Chihuahua",
+ "Myanmar Standard Time": "Asia/Rangoon",
+ "N. Central Asia Standard Time": "Asia/Novosibirsk",
+ "Namibia Standard Time": "Africa/Windhoek",
+ "Nepal Standard Time": "Asia/Katmandu",
+ "New Zealand Standard Time": "Pacific/Auckland",
+ "Newfoundland Standard Time": "America/St_Johns",
+ "Norfolk Standard Time": "Pacific/Norfolk",
+ "North Asia East Standard Time": "Asia/Irkutsk",
+ "North Asia Standard Time": "Asia/Krasnoyarsk",
+ "North Korea Standard Time": "Asia/Pyongyang",
+ "Omsk Standard Time": "Asia/Omsk",
+ "Pacific SA Standard Time": "America/Santiago",
+ "Pacific Standard Time": "America/Los_Angeles",
+ "Pacific Standard Time (Mexico)": "America/Tijuana",
+ "Pakistan Standard Time": "Asia/Karachi",
+ "Paraguay Standard Time": "America/Asuncion",
+ "Romance Standard Time": "Europe/Paris",
+ "Russia Time Zone 10": "Asia/Srednekolymsk",
+ "Russia Time Zone 11": "Asia/Kamchatka",
+ "Russia Time Zone 3": "Europe/Samara",
+ "Russian Standard Time": "Europe/Moscow",
+ "SA Eastern Standard Time": "America/Cayenne",
+ "SA Pacific Standard Time": "America/Bogota",
+ "SA Western Standard Time": "America/La_Paz",
+ "SE Asia Standard Time": "Asia/Bangkok",
+ "Saint Pierre Standard Time": "America/Miquelon",
+ "Sakhalin Standard Time": "Asia/Sakhalin",
+ "Samoa Standard Time": "Pacific/Apia",
+ "Sao Tome Standard Time": "Africa/Sao_Tome",
+ "Saratov Standard Time": "Europe/Saratov",
+ "Singapore Standard Time": "Asia/Singapore",
+ "South Africa Standard Time": "Africa/Johannesburg",
+ "Sri Lanka Standard Time": "Asia/Colombo",
+ "Sudan Standard Time": "Africa/Khartoum",
+ "Syria Standard Time": "Asia/Damascus",
+ "Taipei Standard Time": "Asia/Taipei",
+ "Tasmania Standard Time": "Australia/Hobart",
+ "Tocantins Standard Time": "America/Araguaina",
+ "Tokyo Standard Time": "Asia/Tokyo",
+ "Tomsk Standard Time": "Asia/Tomsk",
+ "Tonga Standard Time": "Pacific/Tongatapu",
+ "Transbaikal Standard Time": "Asia/Chita",
+ "Turkey Standard Time": "Europe/Istanbul",
+ "Turks And Caicos Standard Time": "America/Grand_Turk",
+ "US Eastern Standard Time": "America/Indianapolis",
+ "US Mountain Standard Time": "America/Phoenix",
+ "UTC": "Etc/GMT",
+ "UTC+12": "Etc/GMT-12",
+ "UTC+13": "Etc/GMT-13",
+ "UTC-02": "Etc/GMT+2",
+ "UTC-08": "Etc/GMT+8",
+ "UTC-09": "Etc/GMT+9",
+ "UTC-11": "Etc/GMT+11",
+ "Ulaanbaatar Standard Time": "Asia/Ulaanbaatar",
+ "Venezuela Standard Time": "America/Caracas",
+ "Vladivostok Standard Time": "Asia/Vladivostok",
+ "W. Australia Standard Time": "Australia/Perth",
+ "W. Central Africa Standard Time": "Africa/Lagos",
+ "W. Europe Standard Time": "Europe/Berlin",
+ "W. Mongolia Standard Time": "Asia/Hovd",
+ "West Asia Standard Time": "Asia/Tashkent",
+ "West Bank Standard Time": "Asia/Hebron",
+ "West Pacific Standard Time": "Pacific/Port_Moresby",
+ "Yakutsk Standard Time": "Asia/Yakutsk",
+}
diff --git a/pendulum/tz/exceptions.py b/pendulum/tz/exceptions.py new file mode 100644 index 0000000..d1572f9 --- /dev/null +++ b/pendulum/tz/exceptions.py @@ -0,0 +1,23 @@ +class TimezoneError(ValueError):
+
+ pass
+
+
+class NonExistingTime(TimezoneError):
+
+ message = "The datetime {} does not exist."
+
+ def __init__(self, dt):
+ message = self.message.format(dt)
+
+ super(NonExistingTime, self).__init__(message)
+
+
+class AmbiguousTime(TimezoneError):
+
+ message = "The datetime {} is ambiguous."
+
+ def __init__(self, dt):
+ message = self.message.format(dt)
+
+ super(AmbiguousTime, self).__init__(message)
diff --git a/pendulum/tz/local_timezone.py b/pendulum/tz/local_timezone.py new file mode 100644 index 0000000..756105a --- /dev/null +++ b/pendulum/tz/local_timezone.py @@ -0,0 +1,257 @@ +import os
+import re
+import sys
+
+from contextlib import contextmanager
+from typing import Iterator
+from typing import Optional
+from typing import Union
+
+from .timezone import Timezone
+from .timezone import TimezoneFile
+from .zoneinfo.exceptions import InvalidTimezone
+
+
+try:
+ import _winreg as winreg
+except ImportError:
+ try:
+ import winreg
+ except ImportError:
+ winreg = None
+
+
+_mock_local_timezone = None
+_local_timezone = None
+
+
+def get_local_timezone(): # type: () -> Timezone
+ global _local_timezone
+
+ if _mock_local_timezone is not None:
+ return _mock_local_timezone
+
+ if _local_timezone is None:
+ tz = _get_system_timezone()
+
+ _local_timezone = tz
+
+ return _local_timezone
+
+
+def set_local_timezone(mock=None): # type: (Optional[Union[str, Timezone]]) -> None
+ global _mock_local_timezone
+
+ _mock_local_timezone = mock
+
+
+@contextmanager
+def test_local_timezone(mock): # type: (Timezone) -> Iterator[None]
+ set_local_timezone(mock)
+
+ yield
+
+ set_local_timezone()
+
+
+def _get_system_timezone(): # type: () -> Timezone
+ if sys.platform == "win32":
+ return _get_windows_timezone()
+ elif "darwin" in sys.platform:
+ return _get_darwin_timezone()
+
+ return _get_unix_timezone()
+
+
+def _get_windows_timezone(): # type: () -> Timezone
+ from .data.windows import windows_timezones
+
+ # Windows is special. It has unique time zone names (in several
+ # meanings of the word) available, but unfortunately, they can be
+ # translated to the language of the operating system, so we need to
+ # do a backwards lookup, by going through all time zones and see which
+ # one matches.
+ handle = winreg.ConnectRegistry(None, winreg.HKEY_LOCAL_MACHINE)
+
+ tz_local_key_name = r"SYSTEM\CurrentControlSet\Control\TimeZoneInformation"
+ localtz = winreg.OpenKey(handle, tz_local_key_name)
+
+ timezone_info = {}
+ size = winreg.QueryInfoKey(localtz)[1]
+ for i in range(size):
+ data = winreg.EnumValue(localtz, i)
+ timezone_info[data[0]] = data[1]
+
+ localtz.Close()
+
+ if "TimeZoneKeyName" in timezone_info:
+ # Windows 7 (and Vista?)
+
+ # For some reason this returns a string with loads of NUL bytes at
+ # least on some systems. I don't know if this is a bug somewhere, I
+ # just work around it.
+ tzkeyname = timezone_info["TimeZoneKeyName"].split("\x00", 1)[0]
+ else:
+ # Windows 2000 or XP
+
+ # This is the localized name:
+ tzwin = timezone_info["StandardName"]
+
+ # Open the list of timezones to look up the real name:
+ tz_key_name = r"SOFTWARE\Microsoft\Windows NT\CurrentVersion\Time Zones"
+ tzkey = winreg.OpenKey(handle, tz_key_name)
+
+ # Now, match this value to Time Zone information
+ tzkeyname = None
+ for i in range(winreg.QueryInfoKey(tzkey)[0]):
+ subkey = winreg.EnumKey(tzkey, i)
+ sub = winreg.OpenKey(tzkey, subkey)
+
+ info = {}
+ size = winreg.QueryInfoKey(sub)[1]
+ for i in range(size):
+ data = winreg.EnumValue(sub, i)
+ info[data[0]] = data[1]
+
+ sub.Close()
+ try:
+ if info["Std"] == tzwin:
+ tzkeyname = subkey
+ break
+ except KeyError:
+ # This timezone didn't have proper configuration.
+ # Ignore it.
+ pass
+
+ tzkey.Close()
+ handle.Close()
+
+ if tzkeyname is None:
+ raise LookupError("Can not find Windows timezone configuration")
+
+ timezone = windows_timezones.get(tzkeyname)
+ if timezone is None:
+ # Nope, that didn't work. Try adding "Standard Time",
+ # it seems to work a lot of times:
+ timezone = windows_timezones.get(tzkeyname + " Standard Time")
+
+ # Return what we have.
+ if timezone is None:
+ raise LookupError("Unable to find timezone " + tzkeyname)
+
+ return Timezone(timezone)
+
+
+def _get_darwin_timezone(): # type: () -> Timezone
+ # link will be something like /usr/share/zoneinfo/America/Los_Angeles.
+ link = os.readlink("/etc/localtime")
+ tzname = link[link.rfind("zoneinfo/") + 9 :]
+
+ return Timezone(tzname)
+
+
+def _get_unix_timezone(_root="/"): # type: (str) -> Timezone
+ tzenv = os.environ.get("TZ")
+ if tzenv:
+ try:
+ return _tz_from_env(tzenv)
+ except ValueError:
+ pass
+
+ # Now look for distribution specific configuration files
+ # that contain the timezone name.
+ tzpath = os.path.join(_root, "etc/timezone")
+ if os.path.exists(tzpath):
+ with open(tzpath, "rb") as tzfile:
+ data = tzfile.read()
+
+ # Issue #3 was that /etc/timezone was a zoneinfo file.
+ # That's a misconfiguration, but we need to handle it gracefully:
+ if data[:5] != "TZif2":
+ etctz = data.strip().decode()
+ # Get rid of host definitions and comments:
+ if " " in etctz:
+ etctz, dummy = etctz.split(" ", 1)
+ if "#" in etctz:
+ etctz, dummy = etctz.split("#", 1)
+
+ return Timezone(etctz.replace(" ", "_"))
+
+ # CentOS has a ZONE setting in /etc/sysconfig/clock,
+ # OpenSUSE has a TIMEZONE setting in /etc/sysconfig/clock and
+ # Gentoo has a TIMEZONE setting in /etc/conf.d/clock
+ # We look through these files for a timezone:
+ zone_re = re.compile(r'\s*ZONE\s*=\s*"')
+ timezone_re = re.compile(r'\s*TIMEZONE\s*=\s*"')
+ end_re = re.compile('"')
+
+ for filename in ("etc/sysconfig/clock", "etc/conf.d/clock"):
+ tzpath = os.path.join(_root, filename)
+ if not os.path.exists(tzpath):
+ continue
+
+ with open(tzpath, "rt") as tzfile:
+ data = tzfile.readlines()
+
+ for line in data:
+ # Look for the ZONE= setting.
+ match = zone_re.match(line)
+ if match is None:
+ # No ZONE= setting. Look for the TIMEZONE= setting.
+ match = timezone_re.match(line)
+
+ if match is not None:
+ # Some setting existed
+ line = line[match.end() :]
+ etctz = line[: end_re.search(line).start()]
+
+ parts = list(reversed(etctz.replace(" ", "_").split(os.path.sep)))
+ tzpath = []
+ while parts:
+ tzpath.insert(0, parts.pop(0))
+
+ try:
+ return Timezone(os.path.join(*tzpath))
+ except InvalidTimezone:
+ pass
+
+ # systemd distributions use symlinks that include the zone name,
+ # see manpage of localtime(5) and timedatectl(1)
+ tzpath = os.path.join(_root, "etc", "localtime")
+ if os.path.exists(tzpath) and os.path.islink(tzpath):
+ parts = list(
+ reversed(os.path.realpath(tzpath).replace(" ", "_").split(os.path.sep))
+ )
+ tzpath = []
+ while parts:
+ tzpath.insert(0, parts.pop(0))
+ try:
+ return Timezone(os.path.join(*tzpath))
+ except InvalidTimezone:
+ pass
+
+ # No explicit setting existed. Use localtime
+ for filename in ("etc/localtime", "usr/local/etc/localtime"):
+ tzpath = os.path.join(_root, filename)
+
+ if not os.path.exists(tzpath):
+ continue
+
+ return TimezoneFile(tzpath)
+
+ raise RuntimeError("Unable to find any timezone configuration")
+
+
+def _tz_from_env(tzenv): # type: (str) -> Timezone
+ if tzenv[0] == ":":
+ tzenv = tzenv[1:]
+
+ # TZ specifies a file
+ if os.path.exists(tzenv):
+ return TimezoneFile(tzenv)
+
+ # TZ specifies a zoneinfo zone.
+ try:
+ return Timezone(tzenv)
+ except ValueError:
+ raise
diff --git a/pendulum/tz/timezone.py b/pendulum/tz/timezone.py new file mode 100644 index 0000000..bc94d56 --- /dev/null +++ b/pendulum/tz/timezone.py @@ -0,0 +1,377 @@ +from datetime import datetime
+from datetime import timedelta
+from datetime import tzinfo
+from typing import Optional
+from typing import TypeVar
+from typing import overload
+
+import pendulum
+
+from pendulum.helpers import local_time
+from pendulum.helpers import timestamp
+from pendulum.utils._compat import _HAS_FOLD
+
+from .exceptions import AmbiguousTime
+from .exceptions import NonExistingTime
+from .zoneinfo import read
+from .zoneinfo import read_file
+from .zoneinfo.transition import Transition
+
+
+POST_TRANSITION = "post"
+PRE_TRANSITION = "pre"
+TRANSITION_ERROR = "error"
+
+_datetime = datetime
+_D = TypeVar("_D", bound=datetime)
+
+
+class Timezone(tzinfo):
+ """
+ Represents a named timezone.
+
+ The accepted names are those provided by the IANA time zone database.
+
+ >>> from pendulum.tz.timezone import Timezone
+ >>> tz = Timezone('Europe/Paris')
+ """
+
+ def __init__(self, name, extended=True): # type: (str, bool) -> None
+ tz = read(name, extend=extended)
+
+ self._name = name
+ self._transitions = tz.transitions
+ self._hint = {True: None, False: None}
+
+ @property
+ def name(self): # type: () -> str
+ return self._name
+
+ def convert(self, dt, dst_rule=None): # type: (_D, Optional[str]) -> _D
+ """
+ Converts a datetime in the current timezone.
+
+ If the datetime is naive, it will be normalized.
+
+ >>> from datetime import datetime
+ >>> from pendulum import timezone
+ >>> paris = timezone('Europe/Paris')
+ >>> dt = datetime(2013, 3, 31, 2, 30, fold=1)
+ >>> in_paris = paris.convert(dt)
+ >>> in_paris.isoformat()
+ '2013-03-31T03:30:00+02:00'
+
+ If the datetime is aware, it will be properly converted.
+
+ >>> new_york = timezone('America/New_York')
+ >>> in_new_york = new_york.convert(in_paris)
+ >>> in_new_york.isoformat()
+ '2013-03-30T21:30:00-04:00'
+ """
+ if dt.tzinfo is None:
+ return self._normalize(dt, dst_rule=dst_rule)
+
+ return self._convert(dt)
+
+ def datetime(
+ self, year, month, day, hour=0, minute=0, second=0, microsecond=0
+ ): # type: (int, int, int, int, int, int, int) -> _datetime
+ """
+ Return a normalized datetime for the current timezone.
+ """
+ if _HAS_FOLD:
+ return self.convert(
+ datetime(year, month, day, hour, minute, second, microsecond, fold=1)
+ )
+
+ return self.convert(
+ datetime(year, month, day, hour, minute, second, microsecond),
+ dst_rule=POST_TRANSITION,
+ )
+
+ def _normalize(self, dt, dst_rule=None): # type: (_D, Optional[str]) -> _D
+ sec = timestamp(dt)
+ fold = 0
+ transition = self._lookup_transition(sec)
+
+ if not _HAS_FOLD and dst_rule is None:
+ dst_rule = POST_TRANSITION
+
+ if dst_rule is None:
+ dst_rule = PRE_TRANSITION
+ if dt.fold == 1:
+ dst_rule = POST_TRANSITION
+
+ if sec < transition.local:
+ if transition.is_ambiguous(sec):
+ # Ambiguous time
+ if dst_rule == TRANSITION_ERROR:
+ raise AmbiguousTime(dt)
+
+ # We set the fold attribute for later
+ if dst_rule == POST_TRANSITION:
+ fold = 1
+ elif transition.previous is not None:
+ transition = transition.previous
+
+ if transition:
+ if transition.is_ambiguous(sec):
+ # Ambiguous time
+ if dst_rule == TRANSITION_ERROR:
+ raise AmbiguousTime(dt)
+
+ # We set the fold attribute for later
+ if dst_rule == POST_TRANSITION:
+ fold = 1
+ elif transition.is_missing(sec):
+ # Skipped time
+ if dst_rule == TRANSITION_ERROR:
+ raise NonExistingTime(dt)
+
+ # We adjust accordingly
+ if dst_rule == POST_TRANSITION:
+ sec += transition.fix
+ fold = 1
+ else:
+ sec -= transition.fix
+
+ kwargs = {"tzinfo": self}
+ if _HAS_FOLD or isinstance(dt, pendulum.DateTime):
+ kwargs["fold"] = fold
+
+ return dt.__class__(*local_time(sec, 0, dt.microsecond), **kwargs)
+
+ def _convert(self, dt): # type: (_D) -> _D
+ if dt.tzinfo is self:
+ return self._normalize(dt, dst_rule=POST_TRANSITION)
+
+ if not isinstance(dt.tzinfo, Timezone):
+ return dt.astimezone(self)
+
+ stamp = timestamp(dt)
+
+ if isinstance(dt.tzinfo, FixedTimezone):
+ offset = dt.tzinfo.offset
+ else:
+ transition = dt.tzinfo._lookup_transition(stamp)
+ offset = transition.ttype.offset
+
+ if stamp < transition.local and transition.previous is not None:
+ if (
+ transition.previous.is_ambiguous(stamp)
+ and getattr(dt, "fold", 1) == 0
+ ):
+ pass
+ else:
+ offset = transition.previous.ttype.offset
+
+ stamp -= offset
+
+ transition = self._lookup_transition(stamp, is_utc=True)
+ if stamp < transition.at and transition.previous is not None:
+ transition = transition.previous
+
+ offset = transition.ttype.offset
+ stamp += offset
+ fold = int(not transition.ttype.is_dst())
+
+ kwargs = {"tzinfo": self}
+
+ if _HAS_FOLD or isinstance(dt, pendulum.DateTime):
+ kwargs["fold"] = fold
+
+ return dt.__class__(*local_time(stamp, 0, dt.microsecond), **kwargs)
+
+ def _lookup_transition(
+ self, stamp, is_utc=False
+ ): # type: (int, bool) -> Transition
+ lo, hi = 0, len(self._transitions)
+ hint = self._hint[is_utc]
+ if hint:
+ if stamp == hint[0]:
+ return self._transitions[hint[1]]
+ elif stamp < hint[0]:
+ hi = hint[1]
+ else:
+ lo = hint[1]
+
+ if not is_utc:
+ while lo < hi:
+ mid = (lo + hi) // 2
+ if stamp < self._transitions[mid].to:
+ hi = mid
+ else:
+ lo = mid + 1
+ else:
+ while lo < hi:
+ mid = (lo + hi) // 2
+ if stamp < self._transitions[mid].at:
+ hi = mid
+ else:
+ lo = mid + 1
+
+ if lo >= len(self._transitions):
+ # Beyond last transition
+ lo = len(self._transitions) - 1
+
+ self._hint[is_utc] = (stamp, lo)
+
+ return self._transitions[lo]
+
+ @overload
+ def utcoffset(self, dt): # type: (None) -> None
+ pass
+
+ @overload
+ def utcoffset(self, dt): # type: (_datetime) -> timedelta
+ pass
+
+ def utcoffset(self, dt):
+ if dt is None:
+ return
+
+ transition = self._get_transition(dt)
+
+ return transition.utcoffset()
+
+ def dst(
+ self, dt # type: Optional[_datetime]
+ ): # type: (...) -> Optional[timedelta]
+ if dt is None:
+ return
+
+ transition = self._get_transition(dt)
+
+ if not transition.ttype.is_dst():
+ return timedelta()
+
+ return timedelta(seconds=transition.fix)
+
+ def tzname(self, dt): # type: (Optional[_datetime]) -> Optional[str]
+ if dt is None:
+ return
+
+ transition = self._get_transition(dt)
+
+ return transition.ttype.abbreviation
+
+ def _get_transition(self, dt): # type: (_datetime) -> Transition
+ if dt.tzinfo is not None and dt.tzinfo is not self:
+ dt = dt - dt.utcoffset()
+
+ stamp = timestamp(dt)
+
+ transition = self._lookup_transition(stamp, is_utc=True)
+ else:
+ stamp = timestamp(dt)
+
+ transition = self._lookup_transition(stamp)
+
+ if stamp < transition.local and transition.previous is not None:
+ fold = getattr(dt, "fold", 1)
+ if transition.is_ambiguous(stamp):
+ if fold == 0:
+ transition = transition.previous
+ elif transition.previous.is_ambiguous(stamp) and fold == 0:
+ pass
+ else:
+ transition = transition.previous
+
+ return transition
+
+ def fromutc(self, dt): # type: (_D) -> _D
+ stamp = timestamp(dt)
+
+ transition = self._lookup_transition(stamp, is_utc=True)
+ if stamp < transition.at and transition.previous is not None:
+ transition = transition.previous
+
+ stamp += transition.ttype.offset
+
+ return dt.__class__(*local_time(stamp, 0, dt.microsecond), tzinfo=self)
+
+ def __repr__(self): # type: () -> str
+ return "Timezone('{}')".format(self._name)
+
+ def __getinitargs__(self): # type: () -> tuple
+ return (self._name,)
+
+
+class FixedTimezone(Timezone):
+ def __init__(self, offset, name=None):
+ sign = "-" if offset < 0 else "+"
+
+ minutes = offset / 60
+ hour, minute = divmod(abs(int(minutes)), 60)
+
+ if not name:
+ name = "{0}{1:02d}:{2:02d}".format(sign, hour, minute)
+
+ self._name = name
+ self._offset = offset
+ self._utcoffset = timedelta(seconds=offset)
+
+ @property
+ def offset(self): # type: () -> int
+ return self._offset
+
+ def _normalize(self, dt, dst_rule=None): # type: (_D, Optional[str]) -> _D
+ if _HAS_FOLD:
+ dt = dt.__class__(
+ dt.year,
+ dt.month,
+ dt.day,
+ dt.hour,
+ dt.minute,
+ dt.second,
+ dt.microsecond,
+ tzinfo=self,
+ fold=0,
+ )
+ else:
+ dt = dt.__class__(
+ dt.year,
+ dt.month,
+ dt.day,
+ dt.hour,
+ dt.minute,
+ dt.second,
+ dt.microsecond,
+ tzinfo=self,
+ )
+
+ return dt
+
+ def _convert(self, dt): # type: (_D) -> _D
+ if dt.tzinfo is not self:
+ return dt.astimezone(self)
+
+ return dt
+
+ def utcoffset(self, dt): # type: (Optional[_datetime]) -> timedelta
+ return self._utcoffset
+
+ def dst(self, dt): # type: (Optional[_datetime]) -> timedelta
+ return timedelta()
+
+ def fromutc(self, dt): # type: (_D) -> _D
+ # Use the stdlib datetime's add method to avoid infinite recursion
+ return (datetime.__add__(dt, self._utcoffset)).replace(tzinfo=self)
+
+ def tzname(self, dt): # type: (Optional[_datetime]) -> Optional[str]
+ return self._name
+
+ def __getinitargs__(self): # type: () -> tuple
+ return self._offset, self._name
+
+
+class TimezoneFile(Timezone):
+ def __init__(self, path):
+ tz = read_file(path)
+
+ self._name = ""
+ self._transitions = tz.transitions
+ self._hint = {True: None, False: None}
+
+
+UTC = FixedTimezone(0, "UTC")
diff --git a/pendulum/tz/zoneinfo/__init__.py b/pendulum/tz/zoneinfo/__init__.py new file mode 100644 index 0000000..890351a --- /dev/null +++ b/pendulum/tz/zoneinfo/__init__.py @@ -0,0 +1,16 @@ +from .reader import Reader
+from .timezone import Timezone
+
+
+def read(name, extend=True): # type: (str, bool) -> Timezone
+ """
+ Read the zoneinfo structure for a given timezone name.
+ """
+ return Reader(extend=extend).read_for(name)
+
+
+def read_file(path, extend=True): # type: (str, bool) -> Timezone
+ """
+ Read the zoneinfo structure for a given path.
+ """
+ return Reader(extend=extend).read(path)
diff --git a/pendulum/tz/zoneinfo/exceptions.py b/pendulum/tz/zoneinfo/exceptions.py new file mode 100644 index 0000000..6e29ae2 --- /dev/null +++ b/pendulum/tz/zoneinfo/exceptions.py @@ -0,0 +1,18 @@ +class ZoneinfoError(Exception):
+
+ pass
+
+
+class InvalidZoneinfoFile(ZoneinfoError):
+
+ pass
+
+
+class InvalidTimezone(ZoneinfoError):
+ def __init__(self, name):
+ super(InvalidTimezone, self).__init__('Invalid timezone "{}"'.format(name))
+
+
+class InvalidPosixSpec(ZoneinfoError):
+ def __init__(self, spec):
+ super(InvalidPosixSpec, self).__init__("Invalid POSIX spec: {}".format(spec))
diff --git a/pendulum/tz/zoneinfo/posix_timezone.py b/pendulum/tz/zoneinfo/posix_timezone.py new file mode 100644 index 0000000..a6a7c72 --- /dev/null +++ b/pendulum/tz/zoneinfo/posix_timezone.py @@ -0,0 +1,270 @@ +"""
+Parsing of a POSIX zone spec as described in the TZ part of section 8.3 in
+http://pubs.opengroup.org/onlinepubs/009695399/basedefs/xbd_chap08.html.
+"""
+import re
+
+from typing import Optional
+
+from pendulum.constants import MONTHS_OFFSETS
+from pendulum.constants import SECS_PER_DAY
+
+from .exceptions import InvalidPosixSpec
+
+
+_spec = re.compile(
+ "^"
+ r"(?P<std_abbr><.*?>|[^-+,\d]{3,})"
+ r"(?P<std_offset>([+-])?(\d{1,2})(:\d{2}(:\d{2})?)?)"
+ r"(?P<dst_info>"
+ r" (?P<dst_abbr><.*?>|[^-+,\d]{3,})"
+ r" (?P<dst_offset>([+-])?(\d{1,2})(:\d{2}(:\d{2})?)?)?"
+ r")?"
+ r"(?:,(?P<rules>"
+ r" (?P<dst_start>"
+ r" (?:J\d+|\d+|M\d{1,2}.\d.[0-6])"
+ r" (?:/(?P<dst_start_offset>([+-])?(\d+)(:\d{2}(:\d{2})?)?))?"
+ " )"
+ " ,"
+ r" (?P<dst_end>"
+ r" (?:J\d+|\d+|M\d{1,2}.\d.[0-6])"
+ r" (?:/(?P<dst_end_offset>([+-])?(\d+)(:\d{2}(:\d{2})?)?))?"
+ " )"
+ "))?"
+ "$",
+ re.VERBOSE,
+)
+
+
+def posix_spec(spec): # type: (str) -> PosixTimezone
+ try:
+ return _posix_spec(spec)
+ except ValueError:
+ raise InvalidPosixSpec(spec)
+
+
+def _posix_spec(spec): # type: (str) -> PosixTimezone
+ m = _spec.match(spec)
+ if not m:
+ raise ValueError("Invalid posix spec")
+
+ std_abbr = _parse_abbr(m.group("std_abbr"))
+ std_offset = _parse_offset(m.group("std_offset"))
+
+ dst_abbr = None
+ dst_offset = None
+ if m.group("dst_info"):
+ dst_abbr = _parse_abbr(m.group("dst_abbr"))
+ if m.group("dst_offset"):
+ dst_offset = _parse_offset(m.group("dst_offset"))
+ else:
+ dst_offset = std_offset + 3600
+
+ dst_start = None
+ dst_end = None
+ if m.group("rules"):
+ dst_start = _parse_rule(m.group("dst_start"))
+ dst_end = _parse_rule(m.group("dst_end"))
+
+ return PosixTimezone(std_abbr, std_offset, dst_abbr, dst_offset, dst_start, dst_end)
+
+
+def _parse_abbr(text): # type: (str) -> str
+ return text.lstrip("<").rstrip(">")
+
+
+def _parse_offset(text, sign=-1): # type: (str, int) -> int
+ if text.startswith(("+", "-")):
+ if text.startswith("-"):
+ sign *= -1
+
+ text = text[1:]
+
+ minutes = 0
+ seconds = 0
+
+ parts = text.split(":")
+ hours = int(parts[0])
+
+ if len(parts) > 1:
+ minutes = int(parts[1])
+
+ if len(parts) > 2:
+ seconds = int(parts[2])
+
+ return sign * ((((hours * 60) + minutes) * 60) + seconds)
+
+
+def _parse_rule(rule): # type: (str) -> PosixTransition
+ klass = NPosixTransition
+ args = ()
+
+ if rule.startswith("M"):
+ rule = rule[1:]
+ parts = rule.split(".")
+ month = int(parts[0])
+ week = int(parts[1])
+ day = int(parts[2].split("/")[0])
+
+ args += (month, week, day)
+ klass = MPosixTransition
+ elif rule.startswith("J"):
+ rule = rule[1:]
+ args += (int(rule.split("/")[0]),)
+ klass = JPosixTransition
+ else:
+ args += (int(rule.split("/")[0]),)
+
+ # Checking offset
+ parts = rule.split("/")
+ if len(parts) > 1:
+ offset = _parse_offset(parts[-1], sign=1)
+ else:
+ offset = 7200
+
+ args += (offset,)
+
+ return klass(*args)
+
+
+class PosixTransition(object):
+ def __init__(self, offset): # type: (int) -> None
+ self._offset = offset
+
+ @property
+ def offset(self): # type: () -> int
+ return self._offset
+
+ def trans_offset(self, is_leap, jan1_weekday): # type: (bool, int) -> int
+ raise NotImplementedError()
+
+
+class JPosixTransition(PosixTransition):
+ def __init__(self, day, offset): # type: (int, int) -> None
+ self._day = day
+
+ super(JPosixTransition, self).__init__(offset)
+
+ @property
+ def day(self): # type: () -> int
+ """
+ day of non-leap year [1:365]
+ """
+ return self._day
+
+ def trans_offset(self, is_leap, jan1_weekday): # type: (bool, int) -> int
+ days = self._day
+ if not is_leap or days < MONTHS_OFFSETS[1][3]:
+ days -= 1
+
+ return (days * SECS_PER_DAY) + self._offset
+
+
+class NPosixTransition(PosixTransition):
+ def __init__(self, day, offset): # type: (int, int) -> None
+ self._day = day
+
+ super(NPosixTransition, self).__init__(offset)
+
+ @property
+ def day(self): # type: () -> int
+ """
+ day of year [0:365]
+ """
+ return self._day
+
+ def trans_offset(self, is_leap, jan1_weekday): # type: (bool, int) -> int
+ days = self._day
+
+ return (days * SECS_PER_DAY) + self._offset
+
+
+class MPosixTransition(PosixTransition):
+ def __init__(self, month, week, weekday, offset):
+ # type: (int, int, int, int) -> None
+ self._month = month
+ self._week = week
+ self._weekday = weekday
+
+ super(MPosixTransition, self).__init__(offset)
+
+ @property
+ def month(self): # type: () -> int
+ """
+ month of year [1:12]
+ """
+ return self._month
+
+ @property
+ def week(self): # type: () -> int
+ """
+ week of month [1:5] (5==last)
+ """
+ return self._week
+
+ @property
+ def weekday(self): # type: () -> int
+ """
+ 0==Sun, ..., 6=Sat
+ """
+ return self._weekday
+
+ def trans_offset(self, is_leap, jan1_weekday): # type: (bool, int) -> int
+ last_week = self._week == 5
+ days = MONTHS_OFFSETS[is_leap][self._month + int(last_week)]
+ weekday = (jan1_weekday + days) % 7
+ if last_week:
+ days -= (weekday + 7 - 1 - self._weekday) % 7 + 1
+ else:
+ days += (self._weekday + 7 - weekday) % 7
+ days += (self._week - 1) * 7
+
+ return (days * SECS_PER_DAY) + self._offset
+
+
+class PosixTimezone:
+ """
+ The entirety of a POSIX-string specified time-zone rule.
+
+ The standard abbreviation and offset are always given.
+ """
+
+ def __init__(
+ self,
+ std_abbr, # type: str
+ std_offset, # type: int
+ dst_abbr, # type: Optional[str]
+ dst_offset, # type: Optional[int]
+ dst_start=None, # type: Optional[PosixTransition]
+ dst_end=None, # type: Optional[PosixTransition]
+ ):
+ self._std_abbr = std_abbr
+ self._std_offset = std_offset
+ self._dst_abbr = dst_abbr
+ self._dst_offset = dst_offset
+ self._dst_start = dst_start
+ self._dst_end = dst_end
+
+ @property
+ def std_abbr(self): # type: () -> str
+ return self._std_abbr
+
+ @property
+ def std_offset(self): # type: () -> int
+ return self._std_offset
+
+ @property
+ def dst_abbr(self): # type: () -> Optional[str]
+ return self._dst_abbr
+
+ @property
+ def dst_offset(self): # type: () -> Optional[int]
+ return self._dst_offset
+
+ @property
+ def dst_start(self): # type: () -> Optional[PosixTransition]
+ return self._dst_start
+
+ @property
+ def dst_end(self): # type: () -> Optional[PosixTransition]
+ return self._dst_end
diff --git a/pendulum/tz/zoneinfo/reader.py b/pendulum/tz/zoneinfo/reader.py new file mode 100644 index 0000000..31cb933 --- /dev/null +++ b/pendulum/tz/zoneinfo/reader.py @@ -0,0 +1,224 @@ +import os
+
+from collections import namedtuple
+from struct import unpack
+from typing import IO
+from typing import Any
+from typing import Dict
+from typing import List
+from typing import Optional
+from typing import Tuple
+
+import pytzdata
+
+from pytzdata.exceptions import TimezoneNotFound
+
+from pendulum.utils._compat import PY2
+
+from .exceptions import InvalidTimezone
+from .exceptions import InvalidZoneinfoFile
+from .posix_timezone import PosixTimezone
+from .posix_timezone import posix_spec
+from .timezone import Timezone
+from .transition import Transition
+from .transition_type import TransitionType
+
+
+_offset = namedtuple("offset", "utc_total_offset is_dst abbr_idx")
+
+header = namedtuple(
+ "header",
+ "version " "utclocals " "stdwalls " "leaps " "transitions " "types " "abbr_size",
+)
+
+
+class Reader:
+ """
+ Reads compiled zoneinfo TZif (\0, 2 or 3) files.
+ """
+
+ def __init__(self, extend=True): # type: (bool) -> None
+ self._extend = extend
+
+ def read_for(self, timezone): # type: (str) -> Timezone
+ """
+ Read the zoneinfo structure for a given timezone name.
+
+ :param timezone: The timezone.
+ """
+ try:
+ file_path = pytzdata.tz_path(timezone)
+ except TimezoneNotFound:
+ raise InvalidTimezone(timezone)
+
+ return self.read(file_path)
+
+ def read(self, file_path): # type: (str) -> Timezone
+ """
+ Read a zoneinfo structure from the given path.
+
+ :param file_path: The path of a zoneinfo file.
+ """
+ if not os.path.exists(file_path):
+ raise InvalidZoneinfoFile("The tzinfo file does not exist")
+
+ with open(file_path, "rb") as fd:
+ return self._parse(fd)
+
+ def _check_read(self, fd, nbytes): # type: (...) -> bytes
+ """
+ Reads the given number of bytes from the given file
+ and checks that the correct number of bytes could be read.
+ """
+ result = fd.read(nbytes)
+
+ if (not result and nbytes > 0) or len(result) != nbytes:
+ raise InvalidZoneinfoFile(
+ "Expected {} bytes reading {}, "
+ "but got {}".format(nbytes, fd.name, len(result) if result else 0)
+ )
+
+ if PY2:
+ return bytearray(result)
+
+ return result
+
+ def _parse(self, fd): # type: (...) -> Timezone
+ """
+ Parse a zoneinfo file.
+ """
+ hdr = self._parse_header(fd)
+
+ if hdr.version in (2, 3):
+ # We're skipping the entire v1 file since
+ # at least the same data will be found in TZFile 2.
+ fd.seek(
+ hdr.transitions * 5
+ + hdr.types * 6
+ + hdr.abbr_size
+ + hdr.leaps * 4
+ + hdr.stdwalls
+ + hdr.utclocals,
+ 1,
+ )
+
+ # Parse the second header
+ hdr = self._parse_header(fd)
+
+ if hdr.version != 2 and hdr.version != 3:
+ raise InvalidZoneinfoFile(
+ "Header versions mismatch for file {}".format(fd.name)
+ )
+
+ # Parse the v2 data
+ trans = self._parse_trans_64(fd, hdr.transitions)
+ type_idx = self._parse_type_idx(fd, hdr.transitions)
+ types = self._parse_types(fd, hdr.types)
+ abbrs = self._parse_abbrs(fd, hdr.abbr_size, types)
+
+ fd.seek(hdr.leaps * 8 + hdr.stdwalls + hdr.utclocals, 1)
+
+ trule = self._parse_posix_tz(fd)
+ else:
+ # TZFile v1
+ trans = self._parse_trans_32(fd, hdr.transitions)
+ type_idx = self._parse_type_idx(fd, hdr.transitions)
+ types = self._parse_types(fd, hdr.types)
+ abbrs = self._parse_abbrs(fd, hdr.abbr_size, types)
+ trule = None
+
+ types = [
+ TransitionType(off, is_dst, abbrs[abbr]) for off, is_dst, abbr in types
+ ]
+
+ transitions = []
+ previous = None
+ for trans, idx in zip(trans, type_idx):
+ transition = Transition(trans, types[idx], previous)
+ transitions.append(transition)
+
+ previous = transition
+
+ if not transitions:
+ transitions.append(Transition(0, types[0], None))
+
+ return Timezone(transitions, posix_rule=trule, extended=self._extend)
+
+ def _parse_header(self, fd): # type: (...) -> header
+ buff = self._check_read(fd, 44)
+
+ if buff[:4] != b"TZif":
+ raise InvalidZoneinfoFile(
+ 'The file "{}" has an invalid header.'.format(fd.name)
+ )
+
+ version = {0x00: 1, 0x32: 2, 0x33: 3}.get(buff[4])
+
+ if version is None:
+ raise InvalidZoneinfoFile(
+ 'The file "{}" has an invalid version.'.format(fd.name)
+ )
+
+ hdr = header(version, *unpack(">6l", buff[20:44]))
+
+ return hdr
+
+ def _parse_trans_64(self, fd, n): # type: (IO[Any], int) -> List[int]
+ trans = []
+ for _ in range(n):
+ buff = self._check_read(fd, 8)
+ trans.append(unpack(">q", buff)[0])
+
+ return trans
+
+ def _parse_trans_32(self, fd, n): # type: (IO[Any], int) -> List[int]
+ trans = []
+ for _ in range(n):
+ buff = self._check_read(fd, 4)
+ trans.append(unpack(">i", buff)[0])
+
+ return trans
+
+ def _parse_type_idx(self, fd, n): # type: (IO[Any], int) -> List[int]
+ buff = self._check_read(fd, n)
+
+ return list(unpack("{}B".format(n), buff))
+
+ def _parse_types(
+ self, fd, n
+ ): # type: (IO[Any], int) -> List[Tuple[Any, bool, int]]
+ types = []
+
+ for _ in range(n):
+ buff = self._check_read(fd, 6)
+ offset = unpack(">l", buff[:4])[0]
+ is_dst = buff[4] == 1
+ types.append((offset, is_dst, buff[5]))
+
+ return types
+
+ def _parse_abbrs(
+ self, fd, n, types
+ ): # type: (IO[Any], int, List[Tuple[Any, bool, int]]) -> Dict[int, str]
+ abbrs = {}
+ buff = self._check_read(fd, n)
+
+ for offset, is_dst, idx in types:
+ if idx not in abbrs:
+ abbr = buff[idx : buff.find(b"\0", idx)].decode("utf-8")
+ abbrs[idx] = abbr
+
+ return abbrs
+
+ def _parse_posix_tz(self, fd): # type: (...) -> Optional[PosixTimezone]
+ s = fd.read().decode("utf-8")
+
+ if not s.startswith("\n") or not s.endswith("\n"):
+ raise InvalidZoneinfoFile('Invalid posix rule in file "{}"'.format(fd.name))
+
+ s = s.strip()
+
+ if not s:
+ return
+
+ return posix_spec(s)
diff --git a/pendulum/tz/zoneinfo/timezone.py b/pendulum/tz/zoneinfo/timezone.py new file mode 100644 index 0000000..2147774 --- /dev/null +++ b/pendulum/tz/zoneinfo/timezone.py @@ -0,0 +1,128 @@ +from datetime import datetime
+from typing import List
+from typing import Optional
+
+from pendulum.constants import DAYS_PER_YEAR
+from pendulum.constants import SECS_PER_YEAR
+from pendulum.helpers import is_leap
+from pendulum.helpers import local_time
+from pendulum.helpers import timestamp
+from pendulum.helpers import week_day
+
+from .posix_timezone import PosixTimezone
+from .transition import Transition
+from .transition_type import TransitionType
+
+
+class Timezone:
+ def __init__(
+ self,
+ transitions, # type: List[Transition]
+ posix_rule=None, # type: Optional[PosixTimezone]
+ extended=True, # type: bool
+ ):
+ self._posix_rule = posix_rule
+ self._transitions = transitions
+
+ if extended:
+ self._extends()
+
+ @property
+ def transitions(self): # type: () -> List[Transition]
+ return self._transitions
+
+ @property
+ def posix_rule(self):
+ return self._posix_rule
+
+ def _extends(self):
+ if not self._posix_rule:
+ return
+
+ posix = self._posix_rule
+
+ if not posix.dst_abbr:
+ # std only
+ # The future specification should match the last/default transition
+ ttype = self._transitions[-1].ttype
+ if not self._check_ttype(ttype, posix.std_offset, False, posix.std_abbr):
+ raise ValueError("Posix spec does not match last transition")
+
+ return
+
+ if len(self._transitions) < 2:
+ raise ValueError("Too few transitions for POSIX spec")
+
+ # Extend the transitions for an additional 400 years
+ # using the future specification
+
+ # The future specification should match the last two transitions,
+ # and those transitions should have different is_dst flags.
+ tr0 = self._transitions[-1]
+ tr1 = self._transitions[-2]
+ tt0 = tr0.ttype
+ tt1 = tr1.ttype
+ if tt0.is_dst():
+ dst = tt0
+ std = tt1
+ else:
+ dst = tt1
+ std = tt0
+
+ self._check_ttype(dst, posix.dst_offset, True, posix.dst_abbr)
+ self._check_ttype(std, posix.std_offset, False, posix.std_abbr)
+
+ # Add the transitions to tr1 and back to tr0 for each extra year.
+ last_year = local_time(tr0.local, 0, 0)[0]
+ leap_year = is_leap(last_year)
+ jan1 = datetime(last_year, 1, 1)
+ jan1_time = timestamp(jan1)
+ jan1_weekday = week_day(jan1.year, jan1.month, jan1.day) % 7
+
+ if local_time(tr1.local, 0, 0)[0] != last_year:
+ # Add a single extra transition to align to a calendar year.
+ if tt0.is_dst():
+ pt1 = posix.dst_end
+ else:
+ pt1 = posix.dst_start
+
+ tr1_offset = pt1.trans_offset(leap_year, jan1_weekday)
+ tr = Transition(jan1_time + tr1_offset - tt0.offset, tr1.ttype, tr0)
+ tr0 = tr
+ tr1 = tr0
+ tt0 = tr0.ttype
+ tt1 = tr1.ttype
+
+ if tt0.is_dst():
+ pt1 = posix.dst_end
+ pt0 = posix.dst_start
+ else:
+ pt1 = posix.dst_start
+ pt0 = posix.dst_end
+
+ tr = tr0
+ for year in range(last_year + 1, last_year + 401):
+ jan1_time += SECS_PER_YEAR[leap_year]
+ jan1_weekday = (jan1_weekday + DAYS_PER_YEAR[leap_year]) % 7
+ leap_year = not leap_year and is_leap(year)
+
+ tr1_offset = pt1.trans_offset(leap_year, jan1_weekday)
+ tr = Transition(jan1_time + tr1_offset - tt0.offset, tt1, tr)
+ self._transitions.append(tr)
+
+ tr0_offset = pt0.trans_offset(leap_year, jan1_weekday)
+ tr = Transition(jan1_time + tr0_offset - tt1.offset, tt0, tr)
+ self._transitions.append(tr)
+
+ def _check_ttype(
+ self,
+ ttype, # type: TransitionType
+ offset, # type: int
+ is_dst, # type: bool
+ abbr, # type: str
+ ): # type: (...) -> bool
+ return (
+ ttype.offset == offset
+ and ttype.is_dst() == is_dst
+ and ttype.abbreviation == abbr
+ )
diff --git a/pendulum/tz/zoneinfo/transition.py b/pendulum/tz/zoneinfo/transition.py new file mode 100644 index 0000000..7c6b2f7 --- /dev/null +++ b/pendulum/tz/zoneinfo/transition.py @@ -0,0 +1,77 @@ +from datetime import timedelta
+from typing import Optional
+
+from .transition_type import TransitionType
+
+
+class Transition:
+ def __init__(
+ self,
+ at, # type: int
+ ttype, # type: TransitionType
+ previous, # type: Optional[Transition]
+ ):
+ self._at = at
+
+ if previous:
+ self._local = at + previous.ttype.offset
+ else:
+ self._local = at + ttype.offset
+
+ self._ttype = ttype
+ self._previous = previous
+
+ if self.previous:
+ self._fix = self._ttype.offset - self.previous.ttype.offset
+ else:
+ self._fix = 0
+
+ self._to = self._local + self._fix
+ self._to_utc = self._at + self._fix
+ self._utcoffset = timedelta(seconds=ttype.offset)
+
+ @property
+ def at(self): # type: () -> int
+ return self._at
+
+ @property
+ def local(self): # type: () -> int
+ return self._local
+
+ @property
+ def to(self): # type: () -> int
+ return self._to
+
+ @property
+ def to_utc(self): # type: () -> int
+ return self._to
+
+ @property
+ def ttype(self): # type: () -> TransitionType
+ return self._ttype
+
+ @property
+ def previous(self): # type: () -> Optional[Transition]
+ return self._previous
+
+ @property
+ def fix(self): # type: () -> int
+ return self._fix
+
+ def is_ambiguous(self, stamp): # type: (int) -> bool
+ return self._to <= stamp < self._local
+
+ def is_missing(self, stamp): # type: (int) -> bool
+ return self._local <= stamp < self._to
+
+ def utcoffset(self): # type: () -> timedelta
+ return self._utcoffset
+
+ def __contains__(self, stamp): # type: (int) -> bool
+ if self.previous is None:
+ return stamp < self.local
+
+ return self.previous.local <= stamp < self.local
+
+ def __repr__(self): # type: () -> str
+ return "Transition({} -> {}, {})".format(self._local, self._to, self._ttype)
diff --git a/pendulum/tz/zoneinfo/transition_type.py b/pendulum/tz/zoneinfo/transition_type.py new file mode 100644 index 0000000..dd0a634 --- /dev/null +++ b/pendulum/tz/zoneinfo/transition_type.py @@ -0,0 +1,35 @@ +from datetime import timedelta
+
+from pendulum.utils._compat import PY2
+from pendulum.utils._compat import encode
+
+
+class TransitionType:
+ def __init__(self, offset, is_dst, abbr):
+ self._offset = offset
+ self._is_dst = is_dst
+ self._abbr = abbr
+
+ self._utcoffset = timedelta(seconds=offset)
+
+ @property
+ def offset(self): # type: () -> int
+ return self._offset
+
+ @property
+ def abbreviation(self): # type: () -> str
+ if PY2:
+ return encode(self._abbr)
+
+ return self._abbr
+
+ def is_dst(self): # type: () -> bool
+ return self._is_dst
+
+ def utcoffset(self): # type: () -> timedelta
+ return self._utcoffset
+
+ def __repr__(self): # type: () -> str
+ return "TransitionType({}, {}, {})".format(
+ self._offset, self._is_dst, self._abbr
+ )
|