"""
|
timezone conversion
|
"""
|
cimport cython
|
from cpython.datetime cimport (
|
PyDelta_Check,
|
datetime,
|
datetime_new,
|
import_datetime,
|
timedelta,
|
tzinfo,
|
)
|
from cython cimport Py_ssize_t
|
|
import_datetime()
|
|
import numpy as np
|
import pytz
|
|
cimport numpy as cnp
|
from numpy cimport (
|
int64_t,
|
intp_t,
|
ndarray,
|
uint8_t,
|
)
|
|
cnp.import_array()
|
|
from pandas._libs.tslibs.dtypes cimport (
|
periods_per_day,
|
periods_per_second,
|
)
|
from pandas._libs.tslibs.nattype cimport NPY_NAT
|
from pandas._libs.tslibs.np_datetime cimport (
|
NPY_DATETIMEUNIT,
|
npy_datetimestruct,
|
pandas_datetime_to_datetimestruct,
|
pydatetime_to_dt64,
|
)
|
from pandas._libs.tslibs.timezones cimport (
|
get_dst_info,
|
is_fixed_offset,
|
is_tzlocal,
|
is_utc,
|
is_zoneinfo,
|
utc_stdlib,
|
)
|
|
|
cdef const int64_t[::1] _deltas_placeholder = np.array([], dtype=np.int64)
|
|
|
@cython.freelist(16)
|
@cython.final
|
cdef class Localizer:
|
# cdef:
|
# tzinfo tz
|
# NPY_DATETIMEUNIT _creso
|
# bint use_utc, use_fixed, use_tzlocal, use_dst, use_pytz
|
# ndarray trans
|
# Py_ssize_t ntrans
|
# const int64_t[::1] deltas
|
# int64_t delta
|
# int64_t* tdata
|
|
@cython.initializedcheck(False)
|
@cython.boundscheck(False)
|
def __cinit__(self, tzinfo tz, NPY_DATETIMEUNIT creso):
|
self.tz = tz
|
self._creso = creso
|
self.use_utc = self.use_tzlocal = self.use_fixed = False
|
self.use_dst = self.use_pytz = False
|
self.ntrans = -1 # placeholder
|
self.delta = -1 # placeholder
|
self.deltas = _deltas_placeholder
|
self.tdata = NULL
|
|
if is_utc(tz) or tz is None:
|
self.use_utc = True
|
|
elif is_tzlocal(tz) or is_zoneinfo(tz):
|
self.use_tzlocal = True
|
|
else:
|
trans, deltas, typ = get_dst_info(tz)
|
if creso != NPY_DATETIMEUNIT.NPY_FR_ns:
|
# NB: using floordiv here is implicitly assuming we will
|
# never see trans or deltas that are not an integer number
|
# of seconds.
|
# TODO: avoid these np.array calls
|
if creso == NPY_DATETIMEUNIT.NPY_FR_us:
|
trans = np.array(trans) // 1_000
|
deltas = np.array(deltas) // 1_000
|
elif creso == NPY_DATETIMEUNIT.NPY_FR_ms:
|
trans = np.array(trans) // 1_000_000
|
deltas = np.array(deltas) // 1_000_000
|
elif creso == NPY_DATETIMEUNIT.NPY_FR_s:
|
trans = np.array(trans) // 1_000_000_000
|
deltas = np.array(deltas) // 1_000_000_000
|
else:
|
raise NotImplementedError(creso)
|
|
self.trans = trans
|
self.ntrans = self.trans.shape[0]
|
self.deltas = deltas
|
|
if typ != "pytz" and typ != "dateutil":
|
# static/fixed; in this case we know that len(delta) == 1
|
self.use_fixed = True
|
self.delta = deltas[0]
|
else:
|
self.use_dst = True
|
if typ == "pytz":
|
self.use_pytz = True
|
self.tdata = <int64_t*>cnp.PyArray_DATA(trans)
|
|
@cython.boundscheck(False)
|
cdef int64_t utc_val_to_local_val(
|
self, int64_t utc_val, Py_ssize_t* pos, bint* fold=NULL
|
) except? -1:
|
if self.use_utc:
|
return utc_val
|
elif self.use_tzlocal:
|
return utc_val + _tz_localize_using_tzinfo_api(
|
utc_val, self.tz, to_utc=False, creso=self._creso, fold=fold
|
)
|
elif self.use_fixed:
|
return utc_val + self.delta
|
else:
|
pos[0] = bisect_right_i8(self.tdata, utc_val, self.ntrans) - 1
|
if fold is not NULL:
|
fold[0] = _infer_dateutil_fold(
|
utc_val, self.trans, self.deltas, pos[0]
|
)
|
|
return utc_val + self.deltas[pos[0]]
|
|
|
cdef int64_t tz_localize_to_utc_single(
|
int64_t val,
|
tzinfo tz,
|
object ambiguous=None,
|
object nonexistent=None,
|
NPY_DATETIMEUNIT creso=NPY_DATETIMEUNIT.NPY_FR_ns,
|
) except? -1:
|
"""See tz_localize_to_utc.__doc__"""
|
cdef:
|
int64_t delta
|
int64_t[::1] deltas
|
|
if val == NPY_NAT:
|
return val
|
|
elif is_utc(tz) or tz is None:
|
# TODO: test with non-nano
|
return val
|
|
elif is_tzlocal(tz):
|
return val - _tz_localize_using_tzinfo_api(val, tz, to_utc=True, creso=creso)
|
|
elif is_fixed_offset(tz):
|
_, deltas, _ = get_dst_info(tz)
|
delta = deltas[0]
|
# TODO: de-duplicate with Localizer.__init__
|
if creso != NPY_DATETIMEUNIT.NPY_FR_ns:
|
if creso == NPY_DATETIMEUNIT.NPY_FR_us:
|
delta = delta // 1000
|
elif creso == NPY_DATETIMEUNIT.NPY_FR_ms:
|
delta = delta // 1_000_000
|
elif creso == NPY_DATETIMEUNIT.NPY_FR_s:
|
delta = delta // 1_000_000_000
|
|
return val - delta
|
|
else:
|
return tz_localize_to_utc(
|
np.array([val], dtype="i8"),
|
tz,
|
ambiguous=ambiguous,
|
nonexistent=nonexistent,
|
creso=creso,
|
)[0]
|
|
|
@cython.boundscheck(False)
|
@cython.wraparound(False)
|
def tz_localize_to_utc(
|
ndarray[int64_t] vals,
|
tzinfo tz,
|
object ambiguous=None,
|
object nonexistent=None,
|
NPY_DATETIMEUNIT creso=NPY_DATETIMEUNIT.NPY_FR_ns,
|
):
|
"""
|
Localize tzinfo-naive i8 to given time zone (using pytz). If
|
there are ambiguities in the values, raise AmbiguousTimeError.
|
|
Parameters
|
----------
|
vals : ndarray[int64_t]
|
tz : tzinfo or None
|
ambiguous : str, bool, or arraylike
|
When clocks moved backward due to DST, ambiguous times may arise.
|
For example in Central European Time (UTC+01), when going from 03:00
|
DST to 02:00 non-DST, 02:30:00 local time occurs both at 00:30:00 UTC
|
and at 01:30:00 UTC. In such a situation, the `ambiguous` parameter
|
dictates how ambiguous times should be handled.
|
|
- 'infer' will attempt to infer fall dst-transition hours based on
|
order
|
- bool-ndarray where True signifies a DST time, False signifies a
|
non-DST time (note that this flag is only applicable for ambiguous
|
times, but the array must have the same length as vals)
|
- bool if True, treat all vals as DST. If False, treat them as non-DST
|
- 'NaT' will return NaT where there are ambiguous times
|
|
nonexistent : {None, "NaT", "shift_forward", "shift_backward", "raise", \
|
timedelta-like}
|
How to handle non-existent times when converting wall times to UTC
|
creso : NPY_DATETIMEUNIT, default NPY_FR_ns
|
|
Returns
|
-------
|
localized : ndarray[int64_t]
|
"""
|
cdef:
|
ndarray[uint8_t, cast=True] ambiguous_array
|
Py_ssize_t i, n = vals.shape[0]
|
Py_ssize_t delta_idx_offset, delta_idx
|
int64_t v, left, right, val, new_local, remaining_mins
|
int64_t first_delta, delta
|
int64_t shift_delta = 0
|
ndarray[int64_t] result_a, result_b, dst_hours
|
int64_t[::1] result
|
bint is_zi = False
|
bint infer_dst = False, is_dst = False, fill = False
|
bint shift_forward = False, shift_backward = False
|
bint fill_nonexist = False
|
str stamp
|
Localizer info = Localizer(tz, creso=creso)
|
int64_t pph = periods_per_day(creso) // 24
|
int64_t pps = periods_per_second(creso)
|
npy_datetimestruct dts
|
|
# Vectorized version of DstTzInfo.localize
|
if info.use_utc:
|
return vals.copy()
|
|
# silence false-positive compiler warning
|
ambiguous_array = np.empty(0, dtype=bool)
|
if isinstance(ambiguous, str):
|
if ambiguous == "infer":
|
infer_dst = True
|
elif ambiguous == "NaT":
|
fill = True
|
elif isinstance(ambiguous, bool):
|
is_dst = True
|
if ambiguous:
|
ambiguous_array = np.ones(len(vals), dtype=bool)
|
else:
|
ambiguous_array = np.zeros(len(vals), dtype=bool)
|
elif hasattr(ambiguous, "__iter__"):
|
is_dst = True
|
if len(ambiguous) != len(vals):
|
raise ValueError("Length of ambiguous bool-array must be "
|
"the same size as vals")
|
ambiguous_array = np.asarray(ambiguous, dtype=bool)
|
|
if nonexistent == "NaT":
|
fill_nonexist = True
|
elif nonexistent == "shift_forward":
|
shift_forward = True
|
elif nonexistent == "shift_backward":
|
shift_backward = True
|
elif PyDelta_Check(nonexistent):
|
from .timedeltas import delta_to_nanoseconds
|
shift_delta = delta_to_nanoseconds(nonexistent, reso=creso)
|
elif nonexistent not in ("raise", None):
|
msg = ("nonexistent must be one of {'NaT', 'raise', 'shift_forward', "
|
"shift_backwards} or a timedelta object")
|
raise ValueError(msg)
|
|
result = cnp.PyArray_EMPTY(vals.ndim, vals.shape, cnp.NPY_INT64, 0)
|
|
if info.use_tzlocal and not is_zoneinfo(tz):
|
for i in range(n):
|
v = vals[i]
|
if v == NPY_NAT:
|
result[i] = NPY_NAT
|
else:
|
result[i] = v - _tz_localize_using_tzinfo_api(
|
v, tz, to_utc=True, creso=creso
|
)
|
return result.base # to return underlying ndarray
|
|
elif info.use_fixed:
|
delta = info.delta
|
for i in range(n):
|
v = vals[i]
|
if v == NPY_NAT:
|
result[i] = NPY_NAT
|
else:
|
result[i] = v - delta
|
return result.base # to return underlying ndarray
|
|
# Determine whether each date lies left of the DST transition (store in
|
# result_a) or right of the DST transition (store in result_b)
|
if is_zoneinfo(tz):
|
is_zi = True
|
result_a, result_b =_get_utc_bounds_zoneinfo(
|
vals, tz, creso=creso
|
)
|
else:
|
result_a, result_b =_get_utc_bounds(
|
vals, info.tdata, info.ntrans, info.deltas, creso=creso
|
)
|
|
# silence false-positive compiler warning
|
dst_hours = np.empty(0, dtype=np.int64)
|
if infer_dst:
|
dst_hours = _get_dst_hours(vals, result_a, result_b, creso=creso)
|
|
# Pre-compute delta_idx_offset that will be used if we go down non-existent
|
# paths.
|
# Shift the delta_idx by if the UTC offset of
|
# the target tz is greater than 0 and we're moving forward
|
# or vice versa
|
first_delta = info.deltas[0]
|
if (shift_forward or shift_delta > 0) and first_delta > 0:
|
delta_idx_offset = 1
|
elif (shift_backward or shift_delta < 0) and first_delta < 0:
|
delta_idx_offset = 1
|
else:
|
delta_idx_offset = 0
|
|
for i in range(n):
|
val = vals[i]
|
left = result_a[i]
|
right = result_b[i]
|
if val == NPY_NAT:
|
# TODO: test with non-nano
|
result[i] = val
|
elif left != NPY_NAT and right != NPY_NAT:
|
if left == right:
|
# TODO: test with non-nano
|
result[i] = left
|
else:
|
if infer_dst and dst_hours[i] != NPY_NAT:
|
# TODO: test with non-nano
|
result[i] = dst_hours[i]
|
elif is_dst:
|
if ambiguous_array[i]:
|
result[i] = left
|
else:
|
result[i] = right
|
elif fill:
|
# TODO: test with non-nano; parametrize test_dt_round_tz_ambiguous
|
result[i] = NPY_NAT
|
else:
|
stamp = _render_tstamp(val, creso=creso)
|
raise pytz.AmbiguousTimeError(
|
f"Cannot infer dst time from {stamp}, try using the "
|
"'ambiguous' argument"
|
)
|
elif left != NPY_NAT:
|
result[i] = left
|
elif right != NPY_NAT:
|
# TODO: test with non-nano
|
result[i] = right
|
else:
|
# Handle nonexistent times
|
if shift_forward or shift_backward or shift_delta != 0:
|
# Shift the nonexistent time to the closest existing time
|
remaining_mins = val % pph
|
if shift_delta != 0:
|
# Validate that we don't relocalize on another nonexistent
|
# time
|
if -1 < shift_delta + remaining_mins < pph:
|
raise ValueError(
|
"The provided timedelta will relocalize on a "
|
f"nonexistent time: {nonexistent}"
|
)
|
new_local = val + shift_delta
|
elif shift_forward:
|
new_local = val + (pph - remaining_mins)
|
else:
|
# Subtract 1 since the beginning hour is _inclusive_ of
|
# nonexistent times
|
new_local = val - remaining_mins - 1
|
|
if is_zi:
|
# use the same construction as in _get_utc_bounds_zoneinfo
|
pandas_datetime_to_datetimestruct(new_local, creso, &dts)
|
extra = (dts.ps // 1000) * (pps // 1_000_000_000)
|
|
dt = datetime_new(dts.year, dts.month, dts.day, dts.hour,
|
dts.min, dts.sec, dts.us, None)
|
|
if shift_forward or shift_delta > 0:
|
dt = dt.replace(tzinfo=tz, fold=1)
|
else:
|
dt = dt.replace(tzinfo=tz, fold=0)
|
dt = dt.astimezone(utc_stdlib)
|
dt = dt.replace(tzinfo=None)
|
result[i] = pydatetime_to_dt64(dt, &dts, creso) + extra
|
|
else:
|
delta_idx = bisect_right_i8(info.tdata, new_local, info.ntrans)
|
|
delta_idx = delta_idx - delta_idx_offset
|
result[i] = new_local - info.deltas[delta_idx]
|
elif fill_nonexist:
|
result[i] = NPY_NAT
|
else:
|
stamp = _render_tstamp(val, creso=creso)
|
raise pytz.NonExistentTimeError(stamp)
|
|
return result.base # .base to get underlying ndarray
|
|
|
cdef Py_ssize_t bisect_right_i8(int64_t *data, int64_t val, Py_ssize_t n):
|
# Caller is responsible for checking n > 0
|
# This looks very similar to local_search_right in the ndarray.searchsorted
|
# implementation.
|
cdef:
|
Py_ssize_t pivot, left = 0, right = n
|
|
# edge cases
|
if val > data[n - 1]:
|
return n
|
|
# Caller is responsible for ensuring 'val >= data[0]'. This is
|
# ensured by the fact that 'data' comes from get_dst_info where data[0]
|
# is *always* NPY_NAT+1. If that ever changes, we will need to restore
|
# the following disabled check.
|
# if val < data[0]:
|
# return 0
|
|
while left < right:
|
pivot = left + (right - left) // 2
|
|
if data[pivot] <= val:
|
left = pivot + 1
|
else:
|
right = pivot
|
|
return left
|
|
|
cdef str _render_tstamp(int64_t val, NPY_DATETIMEUNIT creso):
|
""" Helper function to render exception messages"""
|
from pandas._libs.tslibs.timestamps import Timestamp
|
ts = Timestamp._from_value_and_reso(val, creso, None)
|
return str(ts)
|
|
|
cdef _get_utc_bounds(
|
ndarray vals,
|
int64_t* tdata,
|
Py_ssize_t ntrans,
|
const int64_t[::1] deltas,
|
NPY_DATETIMEUNIT creso,
|
):
|
# Determine whether each date lies left of the DST transition (store in
|
# result_a) or right of the DST transition (store in result_b)
|
|
cdef:
|
ndarray result_a, result_b
|
Py_ssize_t i, n = vals.size
|
int64_t val, v_left, v_right
|
Py_ssize_t isl, isr, pos_left, pos_right
|
int64_t ppd = periods_per_day(creso)
|
|
result_a = cnp.PyArray_EMPTY(vals.ndim, vals.shape, cnp.NPY_INT64, 0)
|
result_b = cnp.PyArray_EMPTY(vals.ndim, vals.shape, cnp.NPY_INT64, 0)
|
|
for i in range(n):
|
# This loops resembles the "Find the two best possibilities" block
|
# in pytz's DstTZInfo.localize method.
|
result_a[i] = NPY_NAT
|
result_b[i] = NPY_NAT
|
|
val = vals[i]
|
if val == NPY_NAT:
|
continue
|
|
# TODO: be careful of overflow in val-ppd
|
isl = bisect_right_i8(tdata, val - ppd, ntrans) - 1
|
if isl < 0:
|
isl = 0
|
|
v_left = val - deltas[isl]
|
pos_left = bisect_right_i8(tdata, v_left, ntrans) - 1
|
# timestamp falls to the left side of the DST transition
|
if v_left + deltas[pos_left] == val:
|
result_a[i] = v_left
|
|
# TODO: be careful of overflow in val+ppd
|
isr = bisect_right_i8(tdata, val + ppd, ntrans) - 1
|
if isr < 0:
|
isr = 0
|
|
v_right = val - deltas[isr]
|
pos_right = bisect_right_i8(tdata, v_right, ntrans) - 1
|
# timestamp falls to the right side of the DST transition
|
if v_right + deltas[pos_right] == val:
|
result_b[i] = v_right
|
|
return result_a, result_b
|
|
|
cdef _get_utc_bounds_zoneinfo(ndarray vals, tz, NPY_DATETIMEUNIT creso):
|
"""
|
For each point in 'vals', find the UTC time that it corresponds to if
|
with fold=0 and fold=1. In non-ambiguous cases, these will match.
|
|
Parameters
|
----------
|
vals : ndarray[int64_t]
|
tz : ZoneInfo
|
creso : NPY_DATETIMEUNIT
|
|
Returns
|
-------
|
ndarray[int64_t]
|
ndarray[int64_t]
|
"""
|
cdef:
|
Py_ssize_t i, n = vals.size
|
npy_datetimestruct dts
|
datetime dt, rt, left, right, aware, as_utc
|
int64_t val, pps = periods_per_second(creso)
|
ndarray result_a, result_b
|
|
result_a = cnp.PyArray_EMPTY(vals.ndim, vals.shape, cnp.NPY_INT64, 0)
|
result_b = cnp.PyArray_EMPTY(vals.ndim, vals.shape, cnp.NPY_INT64, 0)
|
|
for i in range(n):
|
val = vals[i]
|
if val == NPY_NAT:
|
result_a[i] = NPY_NAT
|
result_b[i] = NPY_NAT
|
continue
|
|
pandas_datetime_to_datetimestruct(val, creso, &dts)
|
# casting to pydatetime drops nanoseconds etc, which we will
|
# need to re-add later as 'extra'
|
extra = (dts.ps // 1000) * (pps // 1_000_000_000)
|
|
dt = datetime_new(dts.year, dts.month, dts.day, dts.hour,
|
dts.min, dts.sec, dts.us, None)
|
|
aware = dt.replace(tzinfo=tz)
|
as_utc = aware.astimezone(utc_stdlib)
|
rt = as_utc.astimezone(tz)
|
if aware != rt:
|
# AFAICT this means that 'aware' is non-existent
|
# TODO: better way to check this?
|
# mail.python.org/archives/list/datetime-sig@python.org/
|
# thread/57Y3IQAASJOKHX4D27W463XTZIS2NR3M/
|
result_a[i] = NPY_NAT
|
else:
|
left = as_utc.replace(tzinfo=None)
|
result_a[i] = pydatetime_to_dt64(left, &dts, creso) + extra
|
|
aware = dt.replace(fold=1, tzinfo=tz)
|
as_utc = aware.astimezone(utc_stdlib)
|
rt = as_utc.astimezone(tz)
|
if aware != rt:
|
result_b[i] = NPY_NAT
|
else:
|
right = as_utc.replace(tzinfo=None)
|
result_b[i] = pydatetime_to_dt64(right, &dts, creso) + extra
|
|
return result_a, result_b
|
|
|
@cython.boundscheck(False)
|
cdef ndarray[int64_t] _get_dst_hours(
|
# vals, creso only needed here to potential render an exception message
|
const int64_t[:] vals,
|
ndarray[int64_t] result_a,
|
ndarray[int64_t] result_b,
|
NPY_DATETIMEUNIT creso,
|
):
|
cdef:
|
Py_ssize_t i, n = vals.shape[0]
|
ndarray[uint8_t, cast=True] mismatch
|
ndarray[int64_t] delta, dst_hours
|
ndarray[intp_t] switch_idxs, trans_idx, grp, a_idx, b_idx, one_diff
|
list trans_grp
|
intp_t switch_idx
|
int64_t left, right
|
|
dst_hours = cnp.PyArray_EMPTY(result_a.ndim, result_a.shape, cnp.NPY_INT64, 0)
|
dst_hours[:] = NPY_NAT
|
|
mismatch = cnp.PyArray_ZEROS(result_a.ndim, result_a.shape, cnp.NPY_BOOL, 0)
|
|
for i in range(n):
|
left = result_a[i]
|
right = result_b[i]
|
|
# Get the ambiguous hours (given the above, these are the hours
|
# where result_a != result_b and neither of them are NAT)
|
if left != right and left != NPY_NAT and right != NPY_NAT:
|
mismatch[i] = 1
|
|
trans_idx = mismatch.nonzero()[0]
|
|
if trans_idx.size == 1:
|
# see test_tz_localize_to_utc_ambiguous_infer
|
stamp = _render_tstamp(vals[trans_idx[0]], creso=creso)
|
raise pytz.AmbiguousTimeError(
|
f"Cannot infer dst time from {stamp} as there "
|
"are no repeated times"
|
)
|
|
# Split the array into contiguous chunks (where the difference between
|
# indices is 1). These are effectively dst transitions in different
|
# years which is useful for checking that there is not an ambiguous
|
# transition in an individual year.
|
if trans_idx.size > 0:
|
one_diff = np.where(np.diff(trans_idx) != 1)[0] + 1
|
trans_grp = np.array_split(trans_idx, one_diff)
|
|
# Iterate through each day, if there are no hours where the
|
# delta is negative (indicates a repeat of hour) the switch
|
# cannot be inferred
|
for grp in trans_grp:
|
|
delta = np.diff(result_a[grp])
|
if grp.size == 1 or np.all(delta > 0):
|
# see test_tz_localize_to_utc_ambiguous_infer
|
stamp = _render_tstamp(vals[grp[0]], creso=creso)
|
raise pytz.AmbiguousTimeError(stamp)
|
|
# Find the index for the switch and pull from a for dst and b
|
# for standard
|
switch_idxs = (delta <= 0).nonzero()[0]
|
if switch_idxs.size > 1:
|
# see test_tz_localize_to_utc_ambiguous_infer
|
raise pytz.AmbiguousTimeError(
|
f"There are {switch_idxs.size} dst switches when "
|
"there should only be 1."
|
)
|
|
switch_idx = switch_idxs[0] + 1
|
# Pull the only index and adjust
|
a_idx = grp[:switch_idx]
|
b_idx = grp[switch_idx:]
|
dst_hours[grp] = np.hstack((result_a[a_idx], result_b[b_idx]))
|
|
return dst_hours
|
|
|
# ----------------------------------------------------------------------
|
# Timezone Conversion
|
|
cpdef int64_t tz_convert_from_utc_single(
|
int64_t utc_val, tzinfo tz, NPY_DATETIMEUNIT creso=NPY_DATETIMEUNIT.NPY_FR_ns
|
) except? -1:
|
"""
|
Convert the val (in i8) from UTC to tz
|
|
This is a single value version of tz_convert_from_utc.
|
|
Parameters
|
----------
|
utc_val : int64
|
tz : tzinfo
|
creso : NPY_DATETIMEUNIT, default NPY_FR_ns
|
|
Returns
|
-------
|
converted: int64
|
"""
|
cdef:
|
Localizer info = Localizer(tz, creso=creso)
|
Py_ssize_t pos
|
|
# Note: caller is responsible for ensuring utc_val != NPY_NAT
|
return info.utc_val_to_local_val(utc_val, &pos)
|
|
|
# OSError may be thrown by tzlocal on windows at or close to 1970-01-01
|
# see https://github.com/pandas-dev/pandas/pull/37591#issuecomment-720628241
|
cdef int64_t _tz_localize_using_tzinfo_api(
|
int64_t val,
|
tzinfo tz,
|
bint to_utc=True,
|
NPY_DATETIMEUNIT creso=NPY_DATETIMEUNIT.NPY_FR_ns,
|
bint* fold=NULL,
|
) except? -1:
|
"""
|
Convert the i8 representation of a datetime from a general-case timezone to
|
UTC, or vice-versa using the datetime/tzinfo API.
|
|
Private, not intended for use outside of tslibs.tzconversion.
|
|
Parameters
|
----------
|
val : int64_t
|
tz : tzinfo
|
to_utc : bint
|
True if converting _to_ UTC, False if going the other direction.
|
creso : NPY_DATETIMEUNIT
|
fold : bint*, default NULL
|
pointer to fold: whether datetime ends up in a fold or not
|
after adjustment.
|
Only passed with to_utc=False.
|
|
Returns
|
-------
|
delta : int64_t
|
Value to add when converting from utc, subtract when converting to utc.
|
|
Notes
|
-----
|
Sets fold by pointer
|
"""
|
cdef:
|
npy_datetimestruct dts
|
datetime dt
|
int64_t delta
|
timedelta td
|
int64_t pps = periods_per_second(creso)
|
|
pandas_datetime_to_datetimestruct(val, creso, &dts)
|
|
# datetime_new is cython-optimized constructor
|
if not to_utc:
|
# tz.utcoffset only makes sense if datetime
|
# is _wall time_, so if val is a UTC timestamp convert to wall time
|
dt = _astimezone(dts, tz)
|
|
if fold is not NULL:
|
# NB: fold is only passed with to_utc=False
|
fold[0] = dt.fold
|
else:
|
dt = datetime_new(dts.year, dts.month, dts.day, dts.hour,
|
dts.min, dts.sec, dts.us, None)
|
|
td = tz.utcoffset(dt)
|
delta = int(td.total_seconds() * pps)
|
return delta
|
|
|
cdef datetime _astimezone(npy_datetimestruct dts, tzinfo tz):
|
"""
|
Optimized equivalent to:
|
|
dt = datetime(dts.year, dts.month, dts.day, dts.hour,
|
dts.min, dts.sec, dts.us, utc_stdlib)
|
dt = dt.astimezone(tz)
|
|
Derived from the datetime.astimezone implementation at
|
https://github.com/python/cpython/blob/main/Modules/_datetimemodule.c#L6187
|
|
NB: we are assuming tz is not None.
|
"""
|
cdef:
|
datetime result
|
|
result = datetime_new(dts.year, dts.month, dts.day, dts.hour,
|
dts.min, dts.sec, dts.us, tz)
|
return tz.fromutc(result)
|
|
|
# NB: relies on dateutil internals, subject to change.
|
@cython.boundscheck(False)
|
@cython.wraparound(False)
|
cdef bint _infer_dateutil_fold(
|
int64_t value,
|
const int64_t[::1] trans,
|
const int64_t[::1] deltas,
|
Py_ssize_t pos,
|
):
|
"""
|
Infer _TSObject fold property from value by assuming 0 and then setting
|
to 1 if necessary.
|
|
Parameters
|
----------
|
value : int64_t
|
trans : ndarray[int64_t]
|
ndarray of offset transition points in nanoseconds since epoch.
|
deltas : int64_t[:]
|
array of offsets corresponding to transition points in trans.
|
pos : Py_ssize_t
|
Position of the last transition point before taking fold into account.
|
|
Returns
|
-------
|
bint
|
Due to daylight saving time, one wall clock time can occur twice
|
when shifting from summer to winter time; fold describes whether the
|
datetime-like corresponds to the first (0) or the second time (1)
|
the wall clock hits the ambiguous time
|
|
References
|
----------
|
.. [1] "PEP 495 - Local Time Disambiguation"
|
https://www.python.org/dev/peps/pep-0495/#the-fold-attribute
|
"""
|
cdef:
|
bint fold = 0
|
int64_t fold_delta
|
|
if pos > 0:
|
fold_delta = deltas[pos - 1] - deltas[pos]
|
if value - fold_delta < trans[pos]:
|
fold = 1
|
|
return fold
|
