File size: 4,555 Bytes
7885a28 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 |
from datetime import datetime
import numpy as np
import pytest
from pytz import UTC
from pandas._libs.tslibs import (
OutOfBoundsTimedelta,
astype_overflowsafe,
conversion,
iNaT,
timezones,
tz_convert_from_utc,
tzconversion,
)
from pandas import (
Timestamp,
date_range,
)
import pandas._testing as tm
def _compare_utc_to_local(tz_didx):
def f(x):
return tzconversion.tz_convert_from_utc_single(x, tz_didx.tz)
result = tz_convert_from_utc(tz_didx.asi8, tz_didx.tz)
expected = np.vectorize(f)(tz_didx.asi8)
tm.assert_numpy_array_equal(result, expected)
def _compare_local_to_utc(tz_didx, naive_didx):
# Check that tz_localize behaves the same vectorized and pointwise.
err1 = err2 = None
try:
result = tzconversion.tz_localize_to_utc(naive_didx.asi8, tz_didx.tz)
err1 = None
except Exception as err:
err1 = err
try:
expected = naive_didx.map(lambda x: x.tz_localize(tz_didx.tz)).asi8
except Exception as err:
err2 = err
if err1 is not None:
assert type(err1) == type(err2)
else:
assert err2 is None
tm.assert_numpy_array_equal(result, expected)
def test_tz_localize_to_utc_copies():
# GH#46460
arr = np.arange(5, dtype="i8")
result = tz_convert_from_utc(arr, tz=UTC)
tm.assert_numpy_array_equal(result, arr)
assert not np.shares_memory(arr, result)
result = tz_convert_from_utc(arr, tz=None)
tm.assert_numpy_array_equal(result, arr)
assert not np.shares_memory(arr, result)
def test_tz_convert_single_matches_tz_convert_hourly(tz_aware_fixture):
tz = tz_aware_fixture
tz_didx = date_range("2014-03-01", "2015-01-10", freq="h", tz=tz)
naive_didx = date_range("2014-03-01", "2015-01-10", freq="h")
_compare_utc_to_local(tz_didx)
_compare_local_to_utc(tz_didx, naive_didx)
@pytest.mark.parametrize("freq", ["D", "YE"])
def test_tz_convert_single_matches_tz_convert(tz_aware_fixture, freq):
tz = tz_aware_fixture
tz_didx = date_range("2018-01-01", "2020-01-01", freq=freq, tz=tz)
naive_didx = date_range("2018-01-01", "2020-01-01", freq=freq)
_compare_utc_to_local(tz_didx)
_compare_local_to_utc(tz_didx, naive_didx)
@pytest.mark.parametrize(
"arr",
[
pytest.param(np.array([], dtype=np.int64), id="empty"),
pytest.param(np.array([iNaT], dtype=np.int64), id="all_nat"),
],
)
def test_tz_convert_corner(arr):
result = tz_convert_from_utc(arr, timezones.maybe_get_tz("Asia/Tokyo"))
tm.assert_numpy_array_equal(result, arr)
def test_tz_convert_readonly():
# GH#35530
arr = np.array([0], dtype=np.int64)
arr.setflags(write=False)
result = tz_convert_from_utc(arr, UTC)
tm.assert_numpy_array_equal(result, arr)
@pytest.mark.parametrize("copy", [True, False])
@pytest.mark.parametrize("dtype", ["M8[ns]", "M8[s]"])
def test_length_zero_copy(dtype, copy):
arr = np.array([], dtype=dtype)
result = astype_overflowsafe(arr, copy=copy, dtype=np.dtype("M8[ns]"))
if copy:
assert not np.shares_memory(result, arr)
elif arr.dtype == result.dtype:
assert result is arr
else:
assert not np.shares_memory(result, arr)
def test_ensure_datetime64ns_bigendian():
# GH#29684
arr = np.array([np.datetime64(1, "ms")], dtype=">M8[ms]")
result = astype_overflowsafe(arr, dtype=np.dtype("M8[ns]"))
expected = np.array([np.datetime64(1, "ms")], dtype="M8[ns]")
tm.assert_numpy_array_equal(result, expected)
def test_ensure_timedelta64ns_overflows():
arr = np.arange(10).astype("m8[Y]") * 100
msg = r"Cannot convert 300 years to timedelta64\[ns\] without overflow"
with pytest.raises(OutOfBoundsTimedelta, match=msg):
astype_overflowsafe(arr, dtype=np.dtype("m8[ns]"))
class SubDatetime(datetime):
pass
@pytest.mark.parametrize(
"dt, expected",
[
pytest.param(
Timestamp("2000-01-01"), Timestamp("2000-01-01", tz=UTC), id="timestamp"
),
pytest.param(
datetime(2000, 1, 1), datetime(2000, 1, 1, tzinfo=UTC), id="datetime"
),
pytest.param(
SubDatetime(2000, 1, 1),
SubDatetime(2000, 1, 1, tzinfo=UTC),
id="subclassed_datetime",
),
],
)
def test_localize_pydatetime_dt_types(dt, expected):
# GH 25851
# ensure that subclassed datetime works with
# localize_pydatetime
result = conversion.localize_pydatetime(dt, UTC)
assert result == expected
|