def scheduled_times(self, earliest_time='now', latest_time='+1h'):
"""Returns the times when this search is scheduled to run.
By default this method returns the times in the next hour. For different
time ranges, set *earliest_time* and *latest_time*. For example,
for all times in the last day use "earliest_time=-1d" and
"latest_time=now".
:param earliest_time: The earliest time.
:type earliest_time: ``string``
:param latest_time: The latest time.
:type latest_time: ``string``
:return: The list of search times.
"""
response = self.get("scheduled_times",
earliest_time=earliest_time,
latest_time=latest_time)
data = self._load_atom_entry(response)
rec = _parse_atom_entry(data)
times = [datetime.fromtimestamp(int(t))
for t in rec.content.scheduled_times]
return times
python类fromtimestamp()的实例源码
def scheduled_times(self, earliest_time='now', latest_time='+1h'):
"""Returns the times when this search is scheduled to run.
By default this method returns the times in the next hour. For different
time ranges, set *earliest_time* and *latest_time*. For example,
for all times in the last day use "earliest_time=-1d" and
"latest_time=now".
:param earliest_time: The earliest time.
:type earliest_time: ``string``
:param latest_time: The latest time.
:type latest_time: ``string``
:return: The list of search times.
"""
response = self.get("scheduled_times",
earliest_time=earliest_time,
latest_time=latest_time)
data = self._load_atom_entry(response)
rec = _parse_atom_entry(data)
times = [datetime.fromtimestamp(int(t))
for t in rec.content.scheduled_times]
return times
def scheduled_times(self, earliest_time='now', latest_time='+1h'):
"""Returns the times when this search is scheduled to run.
By default this method returns the times in the next hour. For different
time ranges, set *earliest_time* and *latest_time*. For example,
for all times in the last day use "earliest_time=-1d" and
"latest_time=now".
:param earliest_time: The earliest time.
:type earliest_time: ``string``
:param latest_time: The latest time.
:type latest_time: ``string``
:return: The list of search times.
"""
response = self.get("scheduled_times",
earliest_time=earliest_time,
latest_time=latest_time)
data = self._load_atom_entry(response)
rec = _parse_atom_entry(data)
times = [datetime.fromtimestamp(int(t))
for t in rec.content.scheduled_times]
return times
def scheduled_times(self, earliest_time='now', latest_time='+1h'):
"""Returns the times when this search is scheduled to run.
By default this method returns the times in the next hour. For different
time ranges, set *earliest_time* and *latest_time*. For example,
for all times in the last day use "earliest_time=-1d" and
"latest_time=now".
:param earliest_time: The earliest time.
:type earliest_time: ``string``
:param latest_time: The latest time.
:type latest_time: ``string``
:return: The list of search times.
"""
response = self.get("scheduled_times",
earliest_time=earliest_time,
latest_time=latest_time)
data = self._load_atom_entry(response)
rec = _parse_atom_entry(data)
times = [datetime.fromtimestamp(int(t))
for t in rec.content.scheduled_times]
return times
def scheduled_times(self, earliest_time='now', latest_time='+1h'):
"""Returns the times when this search is scheduled to run.
By default this method returns the times in the next hour. For different
time ranges, set *earliest_time* and *latest_time*. For example,
for all times in the last day use "earliest_time=-1d" and
"latest_time=now".
:param earliest_time: The earliest time.
:type earliest_time: ``string``
:param latest_time: The latest time.
:type latest_time: ``string``
:return: The list of search times.
"""
response = self.get("scheduled_times",
earliest_time=earliest_time,
latest_time=latest_time)
data = self._load_atom_entry(response)
rec = _parse_atom_entry(data)
times = [datetime.fromtimestamp(int(t))
for t in rec.content.scheduled_times]
return times
def scheduled_times(self, earliest_time='now', latest_time='+1h'):
"""Returns the times when this search is scheduled to run.
By default this method returns the times in the next hour. For different
time ranges, set *earliest_time* and *latest_time*. For example,
for all times in the last day use "earliest_time=-1d" and
"latest_time=now".
:param earliest_time: The earliest time.
:type earliest_time: ``string``
:param latest_time: The latest time.
:type latest_time: ``string``
:return: The list of search times.
"""
response = self.get("scheduled_times",
earliest_time=earliest_time,
latest_time=latest_time)
data = self._load_atom_entry(response)
rec = _parse_atom_entry(data)
times = [datetime.fromtimestamp(int(t))
for t in rec.content.scheduled_times]
return times
def scheduled_times(self, earliest_time='now', latest_time='+1h'):
"""Returns the times when this search is scheduled to run.
By default this method returns the times in the next hour. For different
time ranges, set *earliest_time* and *latest_time*. For example,
for all times in the last day use "earliest_time=-1d" and
"latest_time=now".
:param earliest_time: The earliest time.
:type earliest_time: ``string``
:param latest_time: The latest time.
:type latest_time: ``string``
:return: The list of search times.
"""
response = self.get("scheduled_times",
earliest_time=earliest_time,
latest_time=latest_time)
data = self._load_atom_entry(response)
rec = _parse_atom_entry(data)
times = [datetime.fromtimestamp(int(t))
for t in rec.content.scheduled_times]
return times
def scheduled_times(self, earliest_time='now', latest_time='+1h'):
"""Returns the times when this search is scheduled to run.
By default this method returns the times in the next hour. For different
time ranges, set *earliest_time* and *latest_time*. For example,
for all times in the last day use "earliest_time=-1d" and
"latest_time=now".
:param earliest_time: The earliest time.
:type earliest_time: ``string``
:param latest_time: The latest time.
:type latest_time: ``string``
:return: The list of search times.
"""
response = self.get("scheduled_times",
earliest_time=earliest_time,
latest_time=latest_time)
data = self._load_atom_entry(response)
rec = _parse_atom_entry(data)
times = [datetime.fromtimestamp(int(t))
for t in rec.content.scheduled_times]
return times
test_timeseries.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 30
收藏 0
点赞 0
评论 0
def test_NaT_methods(self):
# GH 9513
raise_methods = ['astimezone', 'combine', 'ctime', 'dst',
'fromordinal', 'fromtimestamp', 'isocalendar',
'strftime', 'strptime', 'time', 'timestamp',
'timetuple', 'timetz', 'toordinal', 'tzname',
'utcfromtimestamp', 'utcnow', 'utcoffset',
'utctimetuple']
nat_methods = ['date', 'now', 'replace', 'to_datetime', 'today']
nan_methods = ['weekday', 'isoweekday']
for method in raise_methods:
if hasattr(NaT, method):
self.assertRaises(ValueError, getattr(NaT, method))
for method in nan_methods:
if hasattr(NaT, method):
self.assertTrue(np.isnan(getattr(NaT, method)()))
for method in nat_methods:
if hasattr(NaT, method):
self.assertIs(getattr(NaT, method)(), NaT)
# GH 12300
self.assertEqual(NaT.isoformat(), 'NaT')
test_timeseries.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 25
收藏 0
点赞 0
评论 0
def test_class_ops_pytz(self):
tm._skip_if_no_pytz()
from pytz import timezone
def compare(x, y):
self.assertEqual(int(Timestamp(x).value / 1e9),
int(Timestamp(y).value / 1e9))
compare(Timestamp.now(), datetime.now())
compare(Timestamp.now('UTC'), datetime.now(timezone('UTC')))
compare(Timestamp.utcnow(), datetime.utcnow())
compare(Timestamp.today(), datetime.today())
current_time = calendar.timegm(datetime.now().utctimetuple())
compare(Timestamp.utcfromtimestamp(current_time),
datetime.utcfromtimestamp(current_time))
compare(Timestamp.fromtimestamp(current_time),
datetime.fromtimestamp(current_time))
date_component = datetime.utcnow()
time_component = (date_component + timedelta(minutes=10)).time()
compare(Timestamp.combine(date_component, time_component),
datetime.combine(date_component, time_component))
test_timeseries.py 文件源码
项目:PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 26
收藏 0
点赞 0
评论 0
def test_class_ops_dateutil(self):
tm._skip_if_no_dateutil()
from dateutil.tz import tzutc
def compare(x, y):
self.assertEqual(int(np.round(Timestamp(x).value / 1e9)),
int(np.round(Timestamp(y).value / 1e9)))
compare(Timestamp.now(), datetime.now())
compare(Timestamp.now('UTC'), datetime.now(tzutc()))
compare(Timestamp.utcnow(), datetime.utcnow())
compare(Timestamp.today(), datetime.today())
current_time = calendar.timegm(datetime.now().utctimetuple())
compare(Timestamp.utcfromtimestamp(current_time),
datetime.utcfromtimestamp(current_time))
compare(Timestamp.fromtimestamp(current_time),
datetime.fromtimestamp(current_time))
date_component = datetime.utcnow()
time_component = (date_component + timedelta(minutes=10)).time()
compare(Timestamp.combine(date_component, time_component),
datetime.combine(date_component, time_component))
def scheduled_times(self, earliest_time='now', latest_time='+1h'):
"""Returns the times when this search is scheduled to run.
By default this method returns the times in the next hour. For different
time ranges, set *earliest_time* and *latest_time*. For example,
for all times in the last day use "earliest_time=-1d" and
"latest_time=now".
:param earliest_time: The earliest time.
:type earliest_time: ``string``
:param latest_time: The latest time.
:type latest_time: ``string``
:return: The list of search times.
"""
response = self.get("scheduled_times",
earliest_time=earliest_time,
latest_time=latest_time)
data = self._load_atom_entry(response)
rec = _parse_atom_entry(data)
times = [datetime.fromtimestamp(int(t))
for t in rec.content.scheduled_times]
return times
def scheduled_times(self, earliest_time='now', latest_time='+1h'):
"""Returns the times when this search is scheduled to run.
By default this method returns the times in the next hour. For different
time ranges, set *earliest_time* and *latest_time*. For example,
for all times in the last day use "earliest_time=-1d" and
"latest_time=now".
:param earliest_time: The earliest time.
:type earliest_time: ``string``
:param latest_time: The latest time.
:type latest_time: ``string``
:return: The list of search times.
"""
response = self.get("scheduled_times",
earliest_time=earliest_time,
latest_time=latest_time)
data = self._load_atom_entry(response)
rec = _parse_atom_entry(data)
times = [datetime.fromtimestamp(int(t))
for t in rec.content.scheduled_times]
return times
def show_projects(ctx):
"""Show projects."""
from fulmar.scheduler.projectdb import projectdb
projects = projectdb.get_all()
table = []
headers = ['project_name', 'updated_time', 'is_stopped']
for _, project in projects.iteritems():
project_name = project.get('project_name')
update_timestamp = project.get('update_time')
update_time = datetime.datetime.fromtimestamp(update_timestamp).strftime('%Y-%m-%d %H:%M:%S')
is_stopped = 'True' if project.get('is_stopped') else 'False'
table.append([project_name, update_time, is_stopped])
click.echo(tabulate(table, headers, tablefmt="grid", numalign="right"))
def scheduled_times(self, earliest_time='now', latest_time='+1h'):
"""Returns the times when this search is scheduled to run.
By default this method returns the times in the next hour. For different
time ranges, set *earliest_time* and *latest_time*. For example,
for all times in the last day use "earliest_time=-1d" and
"latest_time=now".
:param earliest_time: The earliest time.
:type earliest_time: ``string``
:param latest_time: The latest time.
:type latest_time: ``string``
:return: The list of search times.
"""
response = self.get("scheduled_times",
earliest_time=earliest_time,
latest_time=latest_time)
data = self._load_atom_entry(response)
rec = _parse_atom_entry(data)
times = [datetime.fromtimestamp(int(t))
for t in rec.content.scheduled_times]
return times
def process_gerrit_ssh_issue(self, instance, issue):
ret = {}
ret['review_url'] = issue['url']
if 'shorturl' in instance:
ret['review_url'] = 'http://%s/%s' % (instance['shorturl'],
issue['number'])
ret['header'] = issue['subject']
ret['owner'] = issue['owner']['email']
ret['author'] = ret['owner']
ret['created'] = datetime.fromtimestamp(issue['createdOn'])
ret['modified'] = datetime.fromtimestamp(issue['lastUpdated'])
if 'comments' in issue:
ret['replies'] = self.process_gerrit_ssh_issue_replies(issue['comments'])
else:
ret['replies'] = []
ret['reviewers'] = set(r['author'] for r in ret['replies'])
ret['reviewers'].discard(ret['author'])
return ret
def process_gerrit_ssh_issue(self, instance, issue):
ret = {}
ret['review_url'] = issue['url']
if 'shorturl' in instance:
ret['review_url'] = 'http://%s/%s' % (instance['shorturl'],
issue['number'])
ret['header'] = issue['subject']
ret['owner'] = issue['owner']['email']
ret['author'] = ret['owner']
ret['created'] = datetime.fromtimestamp(issue['createdOn'])
ret['modified'] = datetime.fromtimestamp(issue['lastUpdated'])
if 'comments' in issue:
ret['replies'] = self.process_gerrit_ssh_issue_replies(issue['comments'])
else:
ret['replies'] = []
ret['reviewers'] = set(r['author'] for r in ret['replies'])
ret['reviewers'].discard(ret['author'])
return ret
def sendDocumentToSolr(comment):
dateTimeFormat = '%Y-%m-%dT%H:%M:%SZ'
created_time = datetime.fromtimestamp(long(comment['time'])).strftime(dateTimeFormat)
try:
print "send it to solr"
s.add(
in_reply_to_object_id=objectId,
user_id=comment['fromid'],
name=comment['username'],
like_count=comment['likes'],
id=comment['id'],
created_at=created_time,
text_length_i=len(comment['text']),
text=comment['text']);
except solr.core.SolrException as solrerror:
print "OUCH !! Something bad happened Larry!"
print solrerror
def Do_ALBUM(self):
album = self.api.call("photos.get", uid = self.params["user"], aid = self.params["album"])
photos = []
for cameo in album:
title = None
if cameo["text"]:
title = cameo["text"] + u" (" + unicode(str(datetime.fromtimestamp(int(cameo["created"])))) + u")"
else:
title = unicode(str(datetime.fromtimestamp(int(cameo["created"]))))
title = PrepareString(title)
e = ( title,
cameo.get("src_xxbig") or cameo.get("src_xbig") or cameo.get("src_big") or cameo["src"],
cameo["src"] )
photos.append(e)
for title, url, thumb in photos:
listItem = xbmcgui.ListItem(title, "", thumb, thumb, ) #search history
xbmcplugin.addDirectoryItem(self.handle, url , listItem, False)
def scheduled_times(self, earliest_time='now', latest_time='+1h'):
"""Returns the times when this search is scheduled to run.
By default this method returns the times in the next hour. For different
time ranges, set *earliest_time* and *latest_time*. For example,
for all times in the last day use "earliest_time=-1d" and
"latest_time=now".
:param earliest_time: The earliest time.
:type earliest_time: ``string``
:param latest_time: The latest time.
:type latest_time: ``string``
:return: The list of search times.
"""
response = self.get("scheduled_times",
earliest_time=earliest_time,
latest_time=latest_time)
data = self._load_atom_entry(response)
rec = _parse_atom_entry(data)
times = [datetime.fromtimestamp(int(t))
for t in rec.content.scheduled_times]
return times
def scheduled_times(self, earliest_time='now', latest_time='+1h'):
"""Returns the times when this search is scheduled to run.
By default this method returns the times in the next hour. For different
time ranges, set *earliest_time* and *latest_time*. For example,
for all times in the last day use "earliest_time=-1d" and
"latest_time=now".
:param earliest_time: The earliest time.
:type earliest_time: ``string``
:param latest_time: The latest time.
:type latest_time: ``string``
:return: The list of search times.
"""
response = self.get("scheduled_times",
earliest_time=earliest_time,
latest_time=latest_time)
data = self._load_atom_entry(response)
rec = _parse_atom_entry(data)
times = [datetime.fromtimestamp(int(t))
for t in rec.content.scheduled_times]
return times
def scheduled_times(self, earliest_time='now', latest_time='+1h'):
"""Returns the times when this search is scheduled to run.
By default this method returns the times in the next hour. For different
time ranges, set *earliest_time* and *latest_time*. For example,
for all times in the last day use "earliest_time=-1d" and
"latest_time=now".
:param earliest_time: The earliest time.
:type earliest_time: ``string``
:param latest_time: The latest time.
:type latest_time: ``string``
:return: The list of search times.
"""
response = self.get("scheduled_times",
earliest_time=earliest_time,
latest_time=latest_time)
data = self._load_atom_entry(response)
rec = _parse_atom_entry(data)
times = [datetime.fromtimestamp(int(t))
for t in rec.content.scheduled_times]
return times
def epoch_datetime(epoch):
DT = datetime.fromtimestamp(epoch).strftime(DATETIME_FORMAT)
return "%s" % (DT)
def print_timestamp(timestamp):
try:
#assume, that timestamp is given in seconds with decimal point
ts = float(timestamp)
except ValueError:
return None
return datetime.datetime.fromtimestamp(ts).strftime(settings.EVENT_DATE_FORMAT)
def init_filetime(target_time):
if isinstance(target_time, datetime.datetime):
return _unix_timestamp_to_win_filetime(dt.timestamp())
elif isinstance(target_time, (int, float)):
dt = datetime.fromtimestamp(target_time, datetime.timezone.utc)
return _unix_timestamp_to_win_filetime(dt.timestamp())
elif isinstance(target_time, str):
if target_time.lower() == "now":
unix = datetime.datetime.now().timestamp()
return _unix_timestamp_to_win_filetime(unix)
elif target_time.lower() == "utcnow":
unix = datetime.datetime.now(datetime.timezone.utc).timestamp()
return _unix_timestamp_to_win_filetime(unix)
elif target_time.lower() in ("midnight", "utcmidnight"):
if target_time.lower().startswith("utc"):
dt = datetime.datetime.now(datetime.timezone.utc)
dt = datetime.datetime(dt.year, dt.month, dt.day,
tzinfo=datetime.timezone.utc)
else:
dt = datetime.datetime.now()
dt = datetime.datetime(dt.year, dt.month, dt.day)
return _unix_timestamp_to_win_filetime(dt.timestamp())
else:
raise ValueError("unrecognized time format: " + target_time)
else:
raise TypeError
def test_fromtimestamp(self):
import time
# Try an arbitrary fixed value.
year, month, day = 1999, 9, 19
ts = time.mktime((year, month, day, 0, 0, 0, 0, 0, -1))
d = self.theclass.fromtimestamp(ts)
self.assertEqual(d.year, year)
self.assertEqual(d.month, month)
self.assertEqual(d.day, day)
def test_insane_fromtimestamp(self):
# It's possible that some platform maps time_t to double,
# and that this test will fail there. This test should
# exempt such platforms (provided they return reasonable
# results!).
for insane in -1e200, 1e200:
self.assertRaises(ValueError, self.theclass.fromtimestamp,
insane)
def test_today(self):
import time
# We claim that today() is like fromtimestamp(time.time()), so
# prove it.
for dummy in range(3):
today = self.theclass.today()
ts = time.time()
todayagain = self.theclass.fromtimestamp(ts)
if today == todayagain:
break
# There are several legit reasons that could fail:
# 1. It recently became midnight, between the today() and the
# time() calls.
# 2. The platform time() has such fine resolution that we'll
# never get the same value twice.
# 3. The platform time() has poor resolution, and we just
# happened to call today() right before a resolution quantum
# boundary.
# 4. The system clock got fiddled between calls.
# In any case, wait a little while and try again.
time.sleep(0.1)
# It worked or it didn't. If it didn't, assume it's reason #2, and
# let the test pass if they're within half a second of each other.
if today != todayagain:
self.assertAlmostEqual(todayagain, today,
delta=timedelta(seconds=0.5))
def test_fromtimestamp(self):
import time
ts = time.time()
expected = time.localtime(ts)
got = self.theclass.fromtimestamp(ts)
self.verify_field_equality(expected, got)
def test_microsecond_rounding(self):
# Test whether fromtimestamp "rounds up" floats that are less
# than one microsecond smaller than an integer.
self.assertEqual(self.theclass.fromtimestamp(0.9999999),
self.theclass.fromtimestamp(1))