380 lines
13 KiB
Python
380 lines
13 KiB
Python
|
'''Base classes and helpers for building zone specific tzinfo classes'''
|
||
|
|
|
||
|
|
from datetime import datetime, timedelta, tzinfo
|
||
|
|
from bisect import bisect_right
|
||
|
|
from sets import Set
|
||
|
|
|
||
|
|
import pytz
|
||
|
|
|
||
|
|
__all__ = []
|
||
|
|
|
||
|
|
_timedelta_cache = {}
|
||
|
|
def memorized_timedelta(seconds):
|
||
|
|
'''Create only one instance of each distinct timedelta'''
|
||
|
|
try:
|
||
|
|
return _timedelta_cache[seconds]
|
||
|
|
except KeyError:
|
||
|
|
delta = timedelta(seconds=seconds)
|
||
|
|
_timedelta_cache[seconds] = delta
|
||
|
|
return delta
|
||
|
|
|
||
|
|
_datetime_cache = {}
|
||
|
|
def memorized_datetime(*args):
|
||
|
|
'''Create only one instance of each distinct datetime'''
|
||
|
|
try:
|
||
|
|
return _datetime_cache[args]
|
||
|
|
except KeyError:
|
||
|
|
dt = datetime(*args)
|
||
|
|
_datetime_cache[args] = dt
|
||
|
|
return dt
|
||
|
|
|
||
|
|
_ttinfo_cache = {}
|
||
|
|
def memorized_ttinfo(*args):
|
||
|
|
'''Create only one instance of each distinct tuple'''
|
||
|
|
try:
|
||
|
|
return _ttinfo_cache[args]
|
||
|
|
except KeyError:
|
||
|
|
ttinfo = (
|
||
|
|
memorized_timedelta(args[0]),
|
||
|
|
memorized_timedelta(args[1]),
|
||
|
|
args[2]
|
||
|
|
)
|
||
|
|
_ttinfo_cache[args] = ttinfo
|
||
|
|
return ttinfo
|
||
|
|
|
||
|
|
_notime = memorized_timedelta(0)
|
||
|
|
|
||
|
|
def _to_seconds(td):
|
||
|
|
'''Convert a timedelta to seconds'''
|
||
|
|
return td.seconds + td.days * 24 * 60 * 60
|
||
|
|
|
||
|
|
|
||
|
|
class BaseTzInfo(tzinfo):
|
||
|
|
# Overridden in subclass
|
||
|
|
_utcoffset = None
|
||
|
|
_tzname = None
|
||
|
|
zone = None
|
||
|
|
|
||
|
|
def __str__(self):
|
||
|
|
return self.zone
|
||
|
|
|
||
|
|
|
||
|
|
class StaticTzInfo(BaseTzInfo):
|
||
|
|
'''A timezone that has a constant offset from UTC
|
||
|
|
|
||
|
|
These timezones are rare, as most regions have changed their
|
||
|
|
offset from UTC at some point in their history
|
||
|
|
'''
|
||
|
|
def fromutc(self, dt):
|
||
|
|
'''See datetime.tzinfo.fromutc'''
|
||
|
|
return (dt + self._utcoffset).replace(tzinfo=self)
|
||
|
|
|
||
|
|
def utcoffset(self,dt):
|
||
|
|
'''See datetime.tzinfo.utcoffset'''
|
||
|
|
return self._utcoffset
|
||
|
|
|
||
|
|
def dst(self,dt):
|
||
|
|
'''See datetime.tzinfo.dst'''
|
||
|
|
return _notime
|
||
|
|
|
||
|
|
def tzname(self,dt):
|
||
|
|
'''See datetime.tzinfo.tzname'''
|
||
|
|
return self._tzname
|
||
|
|
|
||
|
|
def localize(self, dt, is_dst=False):
|
||
|
|
'''Convert naive time to local time'''
|
||
|
|
if dt.tzinfo is not None:
|
||
|
|
raise ValueError, 'Not naive datetime (tzinfo is already set)'
|
||
|
|
return dt.replace(tzinfo=self)
|
||
|
|
|
||
|
|
def normalize(self, dt, is_dst=False):
|
||
|
|
'''Correct the timezone information on the given datetime'''
|
||
|
|
if dt.tzinfo is None:
|
||
|
|
raise ValueError, 'Naive time - no tzinfo set'
|
||
|
|
return dt.replace(tzinfo=self)
|
||
|
|
|
||
|
|
def __repr__(self):
|
||
|
|
return '<StaticTzInfo %r>' % (self.zone,)
|
||
|
|
|
||
|
|
def __reduce__(self):
|
||
|
|
# Special pickle to zone remains a singleton and to cope with
|
||
|
|
# database changes.
|
||
|
|
return pytz._p, (self.zone,)
|
||
|
|
|
||
|
|
|
||
|
|
class DstTzInfo(BaseTzInfo):
|
||
|
|
'''A timezone that has a variable offset from UTC
|
||
|
|
|
||
|
|
The offset might change if daylight savings time comes into effect,
|
||
|
|
or at a point in history when the region decides to change their
|
||
|
|
timezone definition.
|
||
|
|
|
||
|
|
'''
|
||
|
|
# Overridden in subclass
|
||
|
|
_utc_transition_times = None # Sorted list of DST transition times in UTC
|
||
|
|
_transition_info = None # [(utcoffset, dstoffset, tzname)] corresponding
|
||
|
|
# to _utc_transition_times entries
|
||
|
|
zone = None
|
||
|
|
|
||
|
|
# Set in __init__
|
||
|
|
_tzinfos = None
|
||
|
|
_dst = None # DST offset
|
||
|
|
|
||
|
|
def __init__(self, _inf=None, _tzinfos=None):
|
||
|
|
if _inf:
|
||
|
|
self._tzinfos = _tzinfos
|
||
|
|
self._utcoffset, self._dst, self._tzname = _inf
|
||
|
|
else:
|
||
|
|
_tzinfos = {}
|
||
|
|
self._tzinfos = _tzinfos
|
||
|
|
self._utcoffset, self._dst, self._tzname = self._transition_info[0]
|
||
|
|
_tzinfos[self._transition_info[0]] = self
|
||
|
|
for inf in self._transition_info[1:]:
|
||
|
|
if not _tzinfos.has_key(inf):
|
||
|
|
_tzinfos[inf] = self.__class__(inf, _tzinfos)
|
||
|
|
|
||
|
|
def fromutc(self, dt):
|
||
|
|
'''See datetime.tzinfo.fromutc'''
|
||
|
|
dt = dt.replace(tzinfo=None)
|
||
|
|
idx = max(0, bisect_right(self._utc_transition_times, dt) - 1)
|
||
|
|
inf = self._transition_info[idx]
|
||
|
|
return (dt + inf[0]).replace(tzinfo=self._tzinfos[inf])
|
||
|
|
|
||
|
|
def normalize(self, dt):
|
||
|
|
'''Correct the timezone information on the given datetime
|
||
|
|
|
||
|
|
If date arithmetic crosses DST boundaries, the tzinfo
|
||
|
|
is not magically adjusted. This method normalizes the
|
||
|
|
tzinfo to the correct one.
|
||
|
|
|
||
|
|
To test, first we need to do some setup
|
||
|
|
|
||
|
|
>>> from pytz import timezone
|
||
|
|
>>> utc = timezone('UTC')
|
||
|
|
>>> eastern = timezone('US/Eastern')
|
||
|
|
>>> fmt = '%Y-%m-%d %H:%M:%S %Z (%z)'
|
||
|
|
|
||
|
|
We next create a datetime right on an end-of-DST transition point,
|
||
|
|
the instant when the wallclocks are wound back one hour.
|
||
|
|
|
||
|
|
>>> utc_dt = datetime(2002, 10, 27, 6, 0, 0, tzinfo=utc)
|
||
|
|
>>> loc_dt = utc_dt.astimezone(eastern)
|
||
|
|
>>> loc_dt.strftime(fmt)
|
||
|
|
'2002-10-27 01:00:00 EST (-0500)'
|
||
|
|
|
||
|
|
Now, if we subtract a few minutes from it, note that the timezone
|
||
|
|
information has not changed.
|
||
|
|
|
||
|
|
>>> before = loc_dt - timedelta(minutes=10)
|
||
|
|
>>> before.strftime(fmt)
|
||
|
|
'2002-10-27 00:50:00 EST (-0500)'
|
||
|
|
|
||
|
|
But we can fix that by calling the normalize method
|
||
|
|
|
||
|
|
>>> before = eastern.normalize(before)
|
||
|
|
>>> before.strftime(fmt)
|
||
|
|
'2002-10-27 01:50:00 EDT (-0400)'
|
||
|
|
|
||
|
|
'''
|
||
|
|
if dt.tzinfo is None:
|
||
|
|
raise ValueError, 'Naive time - no tzinfo set'
|
||
|
|
|
||
|
|
# Convert dt in localtime to UTC
|
||
|
|
offset = dt.tzinfo._utcoffset
|
||
|
|
dt = dt.replace(tzinfo=None)
|
||
|
|
dt = dt - offset
|
||
|
|
# convert it back, and return it
|
||
|
|
return self.fromutc(dt)
|
||
|
|
|
||
|
|
def localize(self, dt, is_dst=False):
|
||
|
|
'''Convert naive time to local time.
|
||
|
|
|
||
|
|
This method should be used to construct localtimes, rather
|
||
|
|
than passing a tzinfo argument to a datetime constructor.
|
||
|
|
|
||
|
|
is_dst is used to determine the correct timezone in the ambigous
|
||
|
|
period at the end of daylight savings time.
|
||
|
|
|
||
|
|
>>> from pytz import timezone
|
||
|
|
>>> fmt = '%Y-%m-%d %H:%M:%S %Z (%z)'
|
||
|
|
>>> amdam = timezone('Europe/Amsterdam')
|
||
|
|
>>> dt = datetime(2004, 10, 31, 2, 0, 0)
|
||
|
|
>>> loc_dt1 = amdam.localize(dt, is_dst=True)
|
||
|
|
>>> loc_dt2 = amdam.localize(dt, is_dst=False)
|
||
|
|
>>> loc_dt1.strftime(fmt)
|
||
|
|
'2004-10-31 02:00:00 CEST (+0200)'
|
||
|
|
>>> loc_dt2.strftime(fmt)
|
||
|
|
'2004-10-31 02:00:00 CET (+0100)'
|
||
|
|
>>> str(loc_dt2 - loc_dt1)
|
||
|
|
'1:00:00'
|
||
|
|
|
||
|
|
Use is_dst=None to raise an AmbiguousTimeError for ambiguous
|
||
|
|
times at the end of daylight savings
|
||
|
|
|
||
|
|
>>> try:
|
||
|
|
... loc_dt1 = amdam.localize(dt, is_dst=None)
|
||
|
|
... except AmbiguousTimeError:
|
||
|
|
... print 'Oops'
|
||
|
|
Oops
|
||
|
|
|
||
|
|
>>> loc_dt1 = amdam.localize(dt, is_dst=None)
|
||
|
|
Traceback (most recent call last):
|
||
|
|
[...]
|
||
|
|
AmbiguousTimeError: 2004-10-31 02:00:00
|
||
|
|
|
||
|
|
is_dst defaults to False
|
||
|
|
|
||
|
|
>>> amdam.localize(dt) == amdam.localize(dt, False)
|
||
|
|
True
|
||
|
|
|
||
|
|
'''
|
||
|
|
if dt.tzinfo is not None:
|
||
|
|
raise ValueError, 'Not naive datetime (tzinfo is already set)'
|
||
|
|
|
||
|
|
# Find the possibly correct timezones. We probably just have one,
|
||
|
|
# but we might end up with two if we are in the end-of-DST
|
||
|
|
# transition period. Or possibly more in some particularly confused
|
||
|
|
# location...
|
||
|
|
possible_loc_dt = Set()
|
||
|
|
for tzinfo in self._tzinfos.values():
|
||
|
|
loc_dt = tzinfo.normalize(dt.replace(tzinfo=tzinfo))
|
||
|
|
if loc_dt.replace(tzinfo=None) == dt:
|
||
|
|
possible_loc_dt.add(loc_dt)
|
||
|
|
|
||
|
|
if len(possible_loc_dt) == 1:
|
||
|
|
return possible_loc_dt.pop()
|
||
|
|
|
||
|
|
# If told to be strict, raise an exception since we have an
|
||
|
|
# ambiguous case
|
||
|
|
if is_dst is None:
|
||
|
|
raise AmbiguousTimeError(dt)
|
||
|
|
|
||
|
|
# Filter out the possiblilities that don't match the requested
|
||
|
|
# is_dst
|
||
|
|
filtered_possible_loc_dt = [
|
||
|
|
p for p in possible_loc_dt
|
||
|
|
if bool(p.tzinfo._dst) == is_dst
|
||
|
|
]
|
||
|
|
|
||
|
|
# Hopefully we only have one possibility left. Return it.
|
||
|
|
if len(filtered_possible_loc_dt) == 1:
|
||
|
|
return filtered_possible_loc_dt[0]
|
||
|
|
|
||
|
|
if len(filtered_possible_loc_dt) == 0:
|
||
|
|
filtered_possible_loc_dt = list(possible_loc_dt)
|
||
|
|
|
||
|
|
# If we get this far, we have in a wierd timezone transition
|
||
|
|
# where the clocks have been wound back but is_dst is the same
|
||
|
|
# in both (eg. Europe/Warsaw 1915 when they switched to CET).
|
||
|
|
# At this point, we just have to guess unless we allow more
|
||
|
|
# hints to be passed in (such as the UTC offset or abbreviation),
|
||
|
|
# but that is just getting silly.
|
||
|
|
#
|
||
|
|
# Choose the earliest (by UTC) applicable timezone.
|
||
|
|
def mycmp(a,b):
|
||
|
|
return cmp(
|
||
|
|
a.replace(tzinfo=None) - a.tzinfo._utcoffset,
|
||
|
|
b.replace(tzinfo=None) - b.tzinfo._utcoffset,
|
||
|
|
)
|
||
|
|
filtered_possible_loc_dt.sort(mycmp)
|
||
|
|
return filtered_possible_loc_dt[0]
|
||
|
|
|
||
|
|
def utcoffset(self, dt):
|
||
|
|
'''See datetime.tzinfo.utcoffset'''
|
||
|
|
return self._utcoffset
|
||
|
|
|
||
|
|
def dst(self, dt):
|
||
|
|
'''See datetime.tzinfo.dst'''
|
||
|
|
return self._dst
|
||
|
|
|
||
|
|
def tzname(self, dt):
|
||
|
|
'''See datetime.tzinfo.tzname'''
|
||
|
|
return self._tzname
|
||
|
|
|
||
|
|
def __repr__(self):
|
||
|
|
if self._dst:
|
||
|
|
dst = 'DST'
|
||
|
|
else:
|
||
|
|
dst = 'STD'
|
||
|
|
if self._utcoffset > _notime:
|
||
|
|
return '<DstTzInfo %r %s+%s %s>' % (
|
||
|
|
self.zone, self._tzname, self._utcoffset, dst
|
||
|
|
)
|
||
|
|
else:
|
||
|
|
return '<DstTzInfo %r %s%s %s>' % (
|
||
|
|
self.zone, self._tzname, self._utcoffset, dst
|
||
|
|
)
|
||
|
|
|
||
|
|
def __reduce__(self):
|
||
|
|
# Special pickle to zone remains a singleton and to cope with
|
||
|
|
# database changes.
|
||
|
|
return pytz._p, (
|
||
|
|
self.zone,
|
||
|
|
_to_seconds(self._utcoffset),
|
||
|
|
_to_seconds(self._dst),
|
||
|
|
self._tzname
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class AmbiguousTimeError(Exception):
|
||
|
|
'''Exception raised when attempting to create an ambiguous wallclock time.
|
||
|
|
|
||
|
|
At the end of a DST transition period, a particular wallclock time will
|
||
|
|
occur twice (once before the clocks are set back, once after). Both
|
||
|
|
possibilities may be correct, unless further information is supplied.
|
||
|
|
|
||
|
|
See DstTzInfo.normalize() for more info
|
||
|
|
'''
|
||
|
|
|
||
|
|
|
||
|
|
def unpickler(zone, utcoffset=None, dstoffset=None, tzname=None):
|
||
|
|
"""Factory function for unpickling pytz tzinfo instances.
|
||
|
|
|
||
|
|
This is shared for both StaticTzInfo and DstTzInfo instances, because
|
||
|
|
database changes could cause a zones implementation to switch between
|
||
|
|
these two base classes and we can't break pickles on a pytz version
|
||
|
|
upgrade.
|
||
|
|
"""
|
||
|
|
# Raises a KeyError if zone no longer exists, which should never happen
|
||
|
|
# and would be a bug.
|
||
|
|
tz = pytz.timezone(zone)
|
||
|
|
|
||
|
|
# A StaticTzInfo - just return it
|
||
|
|
if utcoffset is None:
|
||
|
|
return tz
|
||
|
|
|
||
|
|
# This pickle was created from a DstTzInfo. We need to
|
||
|
|
# determine which of the list of tzinfo instances for this zone
|
||
|
|
# to use in order to restore the state of any datetime instances using
|
||
|
|
# it correctly.
|
||
|
|
utcoffset = memorized_timedelta(utcoffset)
|
||
|
|
dstoffset = memorized_timedelta(dstoffset)
|
||
|
|
try:
|
||
|
|
return tz._tzinfos[(utcoffset, dstoffset, tzname)]
|
||
|
|
except KeyError:
|
||
|
|
# The particular state requested in this timezone no longer exists.
|
||
|
|
# This indicates a corrupt pickle, or the timezone database has been
|
||
|
|
# corrected violently enough to make this particular
|
||
|
|
# (utcoffset,dstoffset) no longer exist in the zone, or the
|
||
|
|
# abbreviation has been changed.
|
||
|
|
pass
|
||
|
|
|
||
|
|
# See if we can find an entry differing only by tzname. Abbreviations
|
||
|
|
# get changed from the initial guess by the database maintainers to
|
||
|
|
# match reality when this information is discovered.
|
||
|
|
for localized_tz in tz._tzinfos.values():
|
||
|
|
if (localized_tz._utcoffset == utcoffset
|
||
|
|
and localized_tz._dst == dstoffset):
|
||
|
|
return localized_tz
|
||
|
|
|
||
|
|
# This (utcoffset, dstoffset) information has been removed from the
|
||
|
|
# zone. Add it back. This might occur when the database maintainers have
|
||
|
|
# corrected incorrect information. datetime instances using this
|
||
|
|
# incorrect information will continue to do so, exactly as they were
|
||
|
|
# before being pickled. This is purely an overly paranoid safety net - I
|
||
|
|
# doubt this will ever been needed in real life.
|
||
|
|
inf = (utcoffset, dstoffset, tzname)
|
||
|
|
tz._tzinfos[inf] = tz.__class__(inf, tz._tzinfos)
|
||
|
|
return tz._tzinfos[inf]
|
||
|
|
|