test_conversion.py 4.5 KB

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