def in_words_from_now(stamp, sep='_', precision='{:0.1f}'):
if stamp is None:
return 'never'
nw = now()
if nw > stamp:
words = ('ago',)
rdate = delta(nw, stamp)
else:
words = ('from', 'now')
rdate = delta(stamp, nw)
if rdate.days > 0 or rdate.weeks > 0 or rdate.months > 0 or rdate.years > 0:
return stamp.astimezone(dateutil.tz.tzlocal()).isoformat()
if rdate.hours > 0:
value = rdate.hours + (rdate.minutes / 60.0)
label = 'hours'
elif rdate.minutes > 0:
value = rdate.minutes + (rdate.seconds / 60.0)
label = 'min'
else:
value = rdate.seconds + (rdate.microseconds / 1000000.0)
label = 'sec'
return sep.join((precision.format(value), label) + words)
python类tz()的实例源码
def datetime_from_timestamp(ts, tz=None, as_utc=False):
# type: (float, dateutil.tz, bool) -> datetime.datetime
"""Convert a timestamp into datetime with offset
:param float ts: timestamp
:param dateutil.tz tz: time zone or local tz if not specified
:param bool as_utc: convert datetime to UTC
:rtype: datetime.datetime
:return: converted timestamp to datetime
"""
if tz is None:
tz = dateutil.tz.tzlocal()
dt = datetime.datetime.fromtimestamp(ts, tz=tz)
if as_utc:
return dt.astimezone(tz=dateutil.tz.tzutc())
else:
return dt
test_timeseries.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 18
收藏 0
点赞 0
评论 0
def test_index_convert_to_datetime_array(self):
tm._skip_if_no_pytz()
def _check_rng(rng):
converted = rng.to_pydatetime()
tm.assertIsInstance(converted, np.ndarray)
for x, stamp in zip(converted, rng):
tm.assertIsInstance(x, datetime)
self.assertEqual(x, stamp.to_pydatetime())
self.assertEqual(x.tzinfo, stamp.tzinfo)
rng = date_range('20090415', '20090519')
rng_eastern = date_range('20090415', '20090519', tz='US/Eastern')
rng_utc = date_range('20090415', '20090519', tz='utc')
_check_rng(rng)
_check_rng(rng_eastern)
_check_rng(rng_utc)
test_timeseries.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 22
收藏 0
点赞 0
评论 0
def test_index_convert_to_datetime_array_explicit_pytz(self):
tm._skip_if_no_pytz()
import pytz
def _check_rng(rng):
converted = rng.to_pydatetime()
tm.assertIsInstance(converted, np.ndarray)
for x, stamp in zip(converted, rng):
tm.assertIsInstance(x, datetime)
self.assertEqual(x, stamp.to_pydatetime())
self.assertEqual(x.tzinfo, stamp.tzinfo)
rng = date_range('20090415', '20090519')
rng_eastern = date_range('20090415', '20090519',
tz=pytz.timezone('US/Eastern'))
rng_utc = date_range('20090415', '20090519', tz=pytz.utc)
_check_rng(rng)
_check_rng(rng_eastern)
_check_rng(rng_utc)
test_timeseries.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 19
收藏 0
点赞 0
评论 0
def test_index_convert_to_datetime_array_dateutil(self):
tm._skip_if_no_dateutil()
import dateutil
def _check_rng(rng):
converted = rng.to_pydatetime()
tm.assertIsInstance(converted, np.ndarray)
for x, stamp in zip(converted, rng):
tm.assertIsInstance(x, datetime)
self.assertEqual(x, stamp.to_pydatetime())
self.assertEqual(x.tzinfo, stamp.tzinfo)
rng = date_range('20090415', '20090519')
rng_eastern = date_range('20090415', '20090519',
tz='dateutil/US/Eastern')
rng_utc = date_range('20090415', '20090519', tz=dateutil.tz.tzutc())
_check_rng(rng)
_check_rng(rng_eastern)
_check_rng(rng_utc)
test_timeseries.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 19
收藏 0
点赞 0
评论 0
def test_dti_constructor_years_only(self):
# GH 6961
for tz in [None, 'UTC', 'Asia/Tokyo', 'dateutil/US/Pacific']:
rng1 = date_range('2014', '2015', freq='M', tz=tz)
expected1 = date_range('2014-01-31', '2014-12-31', freq='M', tz=tz)
rng2 = date_range('2014', '2015', freq='MS', tz=tz)
expected2 = date_range('2014-01-01', '2015-01-01', freq='MS',
tz=tz)
rng3 = date_range('2014', '2020', freq='A', tz=tz)
expected3 = date_range('2014-12-31', '2019-12-31', freq='A', tz=tz)
rng4 = date_range('2014', '2020', freq='AS', tz=tz)
expected4 = date_range('2014-01-01', '2020-01-01', freq='AS',
tz=tz)
for rng, expected in [(rng1, expected1), (rng2, expected2),
(rng3, expected3), (rng4, expected4)]:
tm.assert_index_equal(rng, expected)
test_timeseries.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 18
收藏 0
点赞 0
评论 0
def test_append_concat_tz(self):
# GH 2938
tm._skip_if_no_pytz()
rng = date_range('5/8/2012 1:45', periods=10, freq='5T',
tz='US/Eastern')
rng2 = date_range('5/8/2012 2:35', periods=10, freq='5T',
tz='US/Eastern')
rng3 = date_range('5/8/2012 1:45', periods=20, freq='5T',
tz='US/Eastern')
ts = Series(np.random.randn(len(rng)), rng)
df = DataFrame(np.random.randn(len(rng), 4), index=rng)
ts2 = Series(np.random.randn(len(rng2)), rng2)
df2 = DataFrame(np.random.randn(len(rng2), 4), index=rng2)
result = ts.append(ts2)
result_df = df.append(df2)
self.assertTrue(result.index.equals(rng3))
self.assertTrue(result_df.index.equals(rng3))
appended = rng.append(rng2)
self.assertTrue(appended.equals(rng3))
test_timeseries.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 19
收藏 0
点赞 0
评论 0
def test_append_concat_tz_explicit_pytz(self):
# GH 2938
tm._skip_if_no_pytz()
from pytz import timezone as timezone
rng = date_range('5/8/2012 1:45', periods=10, freq='5T',
tz=timezone('US/Eastern'))
rng2 = date_range('5/8/2012 2:35', periods=10, freq='5T',
tz=timezone('US/Eastern'))
rng3 = date_range('5/8/2012 1:45', periods=20, freq='5T',
tz=timezone('US/Eastern'))
ts = Series(np.random.randn(len(rng)), rng)
df = DataFrame(np.random.randn(len(rng), 4), index=rng)
ts2 = Series(np.random.randn(len(rng2)), rng2)
df2 = DataFrame(np.random.randn(len(rng2), 4), index=rng2)
result = ts.append(ts2)
result_df = df.append(df2)
self.assertTrue(result.index.equals(rng3))
self.assertTrue(result_df.index.equals(rng3))
appended = rng.append(rng2)
self.assertTrue(appended.equals(rng3))
test_timeseries.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 24
收藏 0
点赞 0
评论 0
def test_period_resample_with_local_timezone_dateutil(self):
# GH5430
tm._skip_if_no_dateutil()
import dateutil
local_timezone = 'dateutil/America/Los_Angeles'
start = datetime(year=2013, month=11, day=1, hour=0, minute=0,
tzinfo=dateutil.tz.tzutc())
# 1 day later
end = datetime(year=2013, month=11, day=2, hour=0, minute=0,
tzinfo=dateutil.tz.tzutc())
index = pd.date_range(start, end, freq='H')
series = pd.Series(1, index=index)
series = series.tz_convert(local_timezone)
result = series.resample('D', kind='period').mean()
# Create the expected series
# Index is moved back a day with the timezone conversion from UTC to
# Pacific
expected_index = (pd.period_range(start=start, end=end, freq='D') - 1)
expected = pd.Series(1, index=expected_index)
assert_series_equal(result, expected)
test_timeseries.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 17
收藏 0
点赞 0
评论 0
def test_take(self):
dates = [datetime(2010, 1, 1, 14), datetime(2010, 1, 1, 15),
datetime(2010, 1, 1, 17), datetime(2010, 1, 1, 21)]
for tz in [None, 'US/Eastern', 'Asia/Tokyo']:
idx = DatetimeIndex(start='2010-01-01 09:00',
end='2010-02-01 09:00', freq='H', tz=tz,
name='idx')
expected = DatetimeIndex(dates, freq=None, name='idx', tz=tz)
taken1 = idx.take([5, 6, 8, 12])
taken2 = idx[[5, 6, 8, 12]]
for taken in [taken1, taken2]:
self.assertTrue(taken.equals(expected))
tm.assertIsInstance(taken, DatetimeIndex)
self.assertIsNone(taken.freq)
self.assertEqual(taken.tz, expected.tz)
self.assertEqual(taken.name, expected.name)
test_timeseries.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 19
收藏 0
点赞 0
评论 0
def test_dti_set_index_reindex(self):
# GH 6631
df = DataFrame(np.random.random(6))
idx1 = date_range('2011/01/01', periods=6, freq='M', tz='US/Eastern')
idx2 = date_range('2013', periods=6, freq='A', tz='Asia/Tokyo')
df = df.set_index(idx1)
self.assertTrue(df.index.equals(idx1))
df = df.reindex(idx2)
self.assertTrue(df.index.equals(idx2))
# 11314
# with tz
index = date_range(datetime(2015, 10, 1), datetime(
2015, 10, 1, 23), freq='H', tz='US/Eastern')
df = DataFrame(np.random.randn(24, 1), columns=['a'], index=index)
new_index = date_range(datetime(2015, 10, 2),
datetime(2015, 10, 2, 23),
freq='H', tz='US/Eastern')
# TODO: unused?
result = df.set_index(new_index) # noqa
self.assertEqual(new_index.freq, index.freq)
test_timeseries.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 20
收藏 0
点赞 0
评论 0
def test_cant_compare_tz_naive_w_aware(self):
tm._skip_if_no_pytz()
# #1404
a = Timestamp('3/12/2012')
b = Timestamp('3/12/2012', tz='utc')
self.assertRaises(Exception, a.__eq__, b)
self.assertRaises(Exception, a.__ne__, b)
self.assertRaises(Exception, a.__lt__, b)
self.assertRaises(Exception, a.__gt__, b)
self.assertRaises(Exception, b.__eq__, a)
self.assertRaises(Exception, b.__ne__, a)
self.assertRaises(Exception, b.__lt__, a)
self.assertRaises(Exception, b.__gt__, a)
if sys.version_info < (3, 3):
self.assertRaises(Exception, a.__eq__, b.to_pydatetime())
self.assertRaises(Exception, a.to_pydatetime().__eq__, b)
else:
self.assertFalse(a == b.to_pydatetime())
self.assertFalse(a.to_pydatetime() == b)
test_timeseries.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 29
收藏 0
点赞 0
评论 0
def test_cant_compare_tz_naive_w_aware_explicit_pytz(self):
tm._skip_if_no_pytz()
from pytz import utc
# #1404
a = Timestamp('3/12/2012')
b = Timestamp('3/12/2012', tz=utc)
self.assertRaises(Exception, a.__eq__, b)
self.assertRaises(Exception, a.__ne__, b)
self.assertRaises(Exception, a.__lt__, b)
self.assertRaises(Exception, a.__gt__, b)
self.assertRaises(Exception, b.__eq__, a)
self.assertRaises(Exception, b.__ne__, a)
self.assertRaises(Exception, b.__lt__, a)
self.assertRaises(Exception, b.__gt__, a)
if sys.version_info < (3, 3):
self.assertRaises(Exception, a.__eq__, b.to_pydatetime())
self.assertRaises(Exception, a.to_pydatetime().__eq__, b)
else:
self.assertFalse(a == b.to_pydatetime())
self.assertFalse(a.to_pydatetime() == b)
test_timeseries.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 20
收藏 0
点赞 0
评论 0
def test_cant_compare_tz_naive_w_aware_dateutil(self):
tm._skip_if_no_dateutil()
from dateutil.tz import tzutc
utc = tzutc()
# #1404
a = Timestamp('3/12/2012')
b = Timestamp('3/12/2012', tz=utc)
self.assertRaises(Exception, a.__eq__, b)
self.assertRaises(Exception, a.__ne__, b)
self.assertRaises(Exception, a.__lt__, b)
self.assertRaises(Exception, a.__gt__, b)
self.assertRaises(Exception, b.__eq__, a)
self.assertRaises(Exception, b.__ne__, a)
self.assertRaises(Exception, b.__lt__, a)
self.assertRaises(Exception, b.__gt__, a)
if sys.version_info < (3, 3):
self.assertRaises(Exception, a.__eq__, b.to_pydatetime())
self.assertRaises(Exception, a.to_pydatetime().__eq__, b)
else:
self.assertFalse(a == b.to_pydatetime())
self.assertFalse(a.to_pydatetime() == b)
test_timezones.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 18
收藏 0
点赞 0
评论 0
def test_tz_localize_dti(self):
dti = DatetimeIndex(start='1/1/2005', end='1/1/2005 0:00:30.256',
freq='L')
dti2 = dti.tz_localize(self.tzstr('US/Eastern'))
dti_utc = DatetimeIndex(start='1/1/2005 05:00',
end='1/1/2005 5:00:30.256', freq='L', tz='utc')
self.assert_numpy_array_equal(dti2.values, dti_utc.values)
dti3 = dti2.tz_convert(self.tzstr('US/Pacific'))
self.assert_numpy_array_equal(dti3.values, dti_utc.values)
dti = DatetimeIndex(start='11/6/2011 1:59', end='11/6/2011 2:00',
freq='L')
self.assertRaises(pytz.AmbiguousTimeError, dti.tz_localize,
self.tzstr('US/Eastern'))
dti = DatetimeIndex(start='3/13/2011 1:59', end='3/13/2011 2:00',
freq='L')
self.assertRaises(pytz.NonExistentTimeError, dti.tz_localize,
self.tzstr('US/Eastern'))
test_timezones.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 24
收藏 0
点赞 0
评论 0
def test_utc_box_timestamp_and_localize(self):
rng = date_range('3/11/2012', '3/12/2012', freq='H', tz='utc')
rng_eastern = rng.tz_convert(self.tzstr('US/Eastern'))
tz = self.tz('US/Eastern')
expected = rng[-1].astimezone(tz)
stamp = rng_eastern[-1]
self.assertEqual(stamp, expected)
self.assertEqual(stamp.tzinfo, expected.tzinfo)
# right tzinfo
rng = date_range('3/13/2012', '3/14/2012', freq='H', tz='utc')
rng_eastern = rng.tz_convert(self.tzstr('US/Eastern'))
# test not valid for dateutil timezones.
# self.assertIn('EDT', repr(rng_eastern[0].tzinfo))
self.assertTrue('EDT' in repr(rng_eastern[0].tzinfo) or 'tzfile' in
repr(rng_eastern[0].tzinfo))
test_timezones.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 25
收藏 0
点赞 0
评论 0
def test_with_tz_ambiguous_times(self):
tz = self.tz('US/Eastern')
# March 13, 2011, spring forward, skip from 2 AM to 3 AM
dr = date_range(datetime(2011, 3, 13, 1, 30), periods=3,
freq=datetools.Hour())
self.assertRaises(pytz.NonExistentTimeError, dr.tz_localize, tz)
# after dst transition, it works
dr = date_range(datetime(2011, 3, 13, 3, 30), periods=3,
freq=datetools.Hour(), tz=tz)
# November 6, 2011, fall back, repeat 2 AM hour
dr = date_range(datetime(2011, 11, 6, 1, 30), periods=3,
freq=datetools.Hour())
self.assertRaises(pytz.AmbiguousTimeError, dr.tz_localize, tz)
# UTC is OK
dr = date_range(datetime(2011, 3, 13), periods=48,
freq=datetools.Minute(30), tz=pytz.utc)
test_timezones.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 20
收藏 0
点赞 0
评论 0
def test_infer_tz(self):
eastern = self.tz('US/Eastern')
utc = pytz.utc
_start = datetime(2001, 1, 1)
_end = datetime(2009, 1, 1)
start = self.localize(eastern, _start)
end = self.localize(eastern, _end)
assert (tools._infer_tzinfo(start, end) is self.localize(
eastern, _start).tzinfo)
assert (tools._infer_tzinfo(start, None) is self.localize(
eastern, _start).tzinfo)
assert (tools._infer_tzinfo(None, end) is self.localize(eastern,
_end).tzinfo)
start = utc.localize(_start)
end = utc.localize(_end)
assert (tools._infer_tzinfo(start, end) is utc)
end = self.localize(eastern, _end)
self.assertRaises(Exception, tools._infer_tzinfo, start, end)
self.assertRaises(Exception, tools._infer_tzinfo, end, start)
test_timezones.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 22
收藏 0
点赞 0
评论 0
def test_index_astype_asobject_tzinfos(self):
# #1345
# dates around a dst transition
rng = date_range('2/13/2010', '5/6/2010', tz=self.tzstr('US/Eastern'))
objs = rng.asobject
for i, x in enumerate(objs):
exval = rng[i]
self.assertEqual(x, exval)
self.assertEqual(x.tzinfo, exval.tzinfo)
objs = rng.astype(object)
for i, x in enumerate(objs):
exval = rng[i]
self.assertEqual(x, exval)
self.assertEqual(x.tzinfo, exval.tzinfo)
test_timezones.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 24
收藏 0
点赞 0
评论 0
def test_localized_at_time_between_time(self):
from datetime import time
rng = date_range('4/16/2012', '5/1/2012', freq='H')
ts = Series(np.random.randn(len(rng)), index=rng)
ts_local = ts.tz_localize(self.tzstr('US/Eastern'))
result = ts_local.at_time(time(10, 0))
expected = ts.at_time(time(10, 0)).tz_localize(self.tzstr(
'US/Eastern'))
tm.assert_series_equal(result, expected)
self.assertTrue(self.cmptz(result.index.tz, self.tz('US/Eastern')))
t1, t2 = time(10, 0), time(11, 0)
result = ts_local.between_time(t1, t2)
expected = ts.between_time(t1,
t2).tz_localize(self.tzstr('US/Eastern'))
tm.assert_series_equal(result, expected)
self.assertTrue(self.cmptz(result.index.tz, self.tz('US/Eastern')))
test_timezones.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 18
收藏 0
点赞 0
评论 0
def test_series_frame_tz_localize(self):
rng = date_range('1/1/2011', periods=100, freq='H')
ts = Series(1, index=rng)
result = ts.tz_localize('utc')
self.assertEqual(result.index.tz.zone, 'UTC')
df = DataFrame({'a': 1}, index=rng)
result = df.tz_localize('utc')
expected = DataFrame({'a': 1}, rng.tz_localize('UTC'))
self.assertEqual(result.index.tz.zone, 'UTC')
assert_frame_equal(result, expected)
df = df.T
result = df.tz_localize('utc', axis=1)
self.assertEqual(result.columns.tz.zone, 'UTC')
assert_frame_equal(result, expected.T)
# Can't localize if already tz-aware
rng = date_range('1/1/2011', periods=100, freq='H', tz='utc')
ts = Series(1, index=rng)
tm.assertRaisesRegexp(TypeError, 'Already tz-aware', ts.tz_localize,
'US/Eastern')
test_timezones.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 20
收藏 0
点赞 0
评论 0
def test_series_frame_tz_convert(self):
rng = date_range('1/1/2011', periods=200, freq='D', tz='US/Eastern')
ts = Series(1, index=rng)
result = ts.tz_convert('Europe/Berlin')
self.assertEqual(result.index.tz.zone, 'Europe/Berlin')
df = DataFrame({'a': 1}, index=rng)
result = df.tz_convert('Europe/Berlin')
expected = DataFrame({'a': 1}, rng.tz_convert('Europe/Berlin'))
self.assertEqual(result.index.tz.zone, 'Europe/Berlin')
assert_frame_equal(result, expected)
df = df.T
result = df.tz_convert('Europe/Berlin', axis=1)
self.assertEqual(result.columns.tz.zone, 'Europe/Berlin')
assert_frame_equal(result, expected.T)
# can't convert tz-naive
rng = date_range('1/1/2011', periods=200, freq='D')
ts = Series(1, index=rng)
tm.assertRaisesRegexp(TypeError, "Cannot convert tz-naive",
ts.tz_convert, 'US/Eastern')
test_timezones.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 24
收藏 0
点赞 0
评论 0
def test_append_aware_naive(self):
rng1 = date_range('1/1/2011 01:00', periods=1, freq='H')
rng2 = date_range('1/1/2011 02:00', periods=1, freq='H',
tz='US/Eastern')
ts1 = Series(np.random.randn(len(rng1)), index=rng1)
ts2 = Series(np.random.randn(len(rng2)), index=rng2)
ts_result = ts1.append(ts2)
self.assertTrue(ts_result.index.equals(ts1.index.asobject.append(
ts2.index.asobject)))
# mixed
rng1 = date_range('1/1/2011 01:00', periods=1, freq='H')
rng2 = lrange(100)
ts1 = Series(np.random.randn(len(rng1)), index=rng1)
ts2 = Series(np.random.randn(len(rng2)), index=rng2)
ts_result = ts1.append(ts2)
self.assertTrue(ts_result.index.equals(ts1.index.asobject.append(
ts2.index)))
test_timezones.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 18
收藏 0
点赞 0
评论 0
def test_equal_join_ensure_utc(self):
rng = date_range('1/1/2011', periods=10, freq='H', tz='US/Eastern')
ts = Series(np.random.randn(len(rng)), index=rng)
ts_moscow = ts.tz_convert('Europe/Moscow')
result = ts + ts_moscow
self.assertIs(result.index.tz, pytz.utc)
result = ts_moscow + ts
self.assertIs(result.index.tz, pytz.utc)
df = DataFrame({'a': ts})
df_moscow = df.tz_convert('Europe/Moscow')
result = df + df_moscow
self.assertIs(result.index.tz, pytz.utc)
result = df_moscow + df
self.assertIs(result.index.tz, pytz.utc)
test_timezones.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 22
收藏 0
点赞 0
评论 0
def test_arith_utc_convert(self):
rng = date_range('1/1/2011', periods=100, freq='H', tz='utc')
perm = np.random.permutation(100)[:90]
ts1 = Series(np.random.randn(90),
index=rng.take(perm).tz_convert('US/Eastern'))
perm = np.random.permutation(100)[:90]
ts2 = Series(np.random.randn(90),
index=rng.take(perm).tz_convert('Europe/Berlin'))
result = ts1 + ts2
uts1 = ts1.tz_convert('utc')
uts2 = ts2.tz_convert('utc')
expected = uts1 + uts2
self.assertEqual(result.index.tz, pytz.UTC)
tm.assert_series_equal(result, expected)
test_timezones.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 25
收藏 0
点赞 0
评论 0
def test_tzaware_offset(self):
dates = date_range('2012-11-01', periods=3, tz='US/Pacific')
offset = dates + offsets.Hour(5)
self.assertEqual(dates[0] + offsets.Hour(5), offset[0])
# GH 6818
for tz in ['UTC', 'US/Pacific', 'Asia/Tokyo']:
dates = date_range('2010-11-01 00:00', periods=3, tz=tz, freq='H')
expected = DatetimeIndex(['2010-11-01 05:00', '2010-11-01 06:00',
'2010-11-01 07:00'], freq='H', tz=tz)
offset = dates + offsets.Hour(5)
self.assertTrue(offset.equals(expected))
offset = dates + np.timedelta64(5, 'h')
self.assertTrue(offset.equals(expected))
offset = dates + timedelta(hours=5)
self.assertTrue(offset.equals(expected))
def create_curtailment_publish(self, current_time_str, device_name, meta):
try:
headers = {
"Date": current_time_str,
"min_compatible_version": "3.0",
"MessageType": "Control"
}
subdevices = self.curtailment.get_device(device_name).command_status.keys()
for subdevice in subdevices:
currently_curtailed = self.curtailment.get_device(device_name).currently_curtailed[subdevice]
curtailment_topic = "/".join([self.update_base_topic, device_name[0], subdevice])
curtailment_status = "Active" if currently_curtailed else "Inactive"
curtailment_message = [
{
"DeviceState": curtailment_status
},
{
"DeviceState": {"tz": "US/Pacific", "type": "string"}
}
]
self.vip.pubsub.publish('pubsub', curtailment_topic, headers=headers, message=curtailment_message).get(timeout=15.0)
except:
_log.debug("Unable to publish device/subdevice curtailment status message.")
def test_vtimezone_creation(self):
tzs = dateutil.tz.tzical("test_files/timezones.ics")
pacific = icalendar.TimezoneComponent(tzs.get('US/Pacific'))
self.assertEqual(
str(pacific),
"<VTIMEZONE | <TZID{}US/Pacific>>"
)
santiago = icalendar.TimezoneComponent(tzs.get('Santiago'))
self.assertEqual(
str(santiago),
"<VTIMEZONE | <TZID{}Santiago>>"
)
for year in range(2001, 2010):
for month in (2, 9):
dt = datetime.datetime(year, month, 15, tzinfo = tzs.get('Santiago'))
#if dt.replace(tzinfo=tzs.get('Santiago')) != dt:
self.assertTrue(dt.replace(tzinfo=tzs.get('Santiago')), dt)
def test_scratchbuild(self):
"""
CreateCalendar 2.0 format from scratch
"""
test_cal = get_test_file("simple_2_0_test.ics")
cal = base.newFromBehavior('vcalendar', '2.0')
cal.add('vevent')
cal.vevent.add('dtstart').value = datetime.datetime(2006, 5, 9)
cal.vevent.add('description').value = "Test event"
cal.vevent.add('created').value = \
datetime.datetime(2006, 1, 1, 10,
tzinfo=dateutil.tz.tzical(
"test_files/timezones.ics").get('US/Pacific'))
cal.vevent.add('uid').value = "Not very random UID"
cal.vevent.add('dtstamp').value = datetime.datetime(2017, 6, 26, 0, tzinfo=tzutc())
# Note we're normalizing line endings, because no one got time for that.
self.assertEqual(
cal.serialize().replace('\r\n', '\n'),
test_cal.replace('\r\n', '\n')
)
def test_vtimezone_creation(self):
"""
Test timezones
"""
tzs = dateutil.tz.tzical("test_files/timezones.ics")
pacific = icalendar.TimezoneComponent(tzs.get('US/Pacific'))
self.assertEqual(
str(pacific),
"<VTIMEZONE | <TZID{}US/Pacific>>"
)
santiago = icalendar.TimezoneComponent(tzs.get('Santiago'))
self.assertEqual(
str(santiago),
"<VTIMEZONE | <TZID{}Santiago>>"
)
for year in range(2001, 2010):
for month in (2, 9):
dt = datetime.datetime(year, month, 15,
tzinfo=tzs.get('Santiago'))
self.assertTrue(dt.replace(tzinfo=tzs.get('Santiago')), dt)