s464967/Inzynierka
2
1
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
master
Inzynierka/Lib/site-packages/pandas/_libs/tslibs/tzconversion.pyx

817 lines
26 KiB
Cython
Raw Permalink Normal View History

first commit
2023-06-02 12:51:02 +02:00
"""
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
Reference in New Issue Copy Permalink