def restart(bot, update):
if not from_admin(update):
return
import os
import time
import sys
update.message.reply_text('Restarting {}…'.format(bot.first_name.title()))
time.sleep(0.2)
os.execl(sys.executable, sys.executable, *sys.argv)
python类sleep()的实例源码
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.
self.assertTrue(today == todayagain or
abs(todayagain - today) < timedelta(seconds=0.5))
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_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_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.
self.assertTrue(today == todayagain or
abs(todayagain - today) < timedelta(seconds=0.5))
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_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 __lsdir(self,path):#
if DEBUG: print "ls dir"
ss = path.split('/')[1]
subdir = '/'+'/'.join(path.split('/')[2:])
dirents = ['.', '..']
#if path == "/" and (self.root.strip('/') == self.BDIRS['loc'].strip('/') or self.root == self.BDIRS['ext'].strip('/')):
if path == "/" and (self.root in self.BDIRS.values()):
self.root = self.BDIRS['loc']
dirents.extend(self.snapshots.keys())
elif ss in self.snapshots or path == "./":
dirents.extend(os.listdir(self.__realpath(path)))
else:
self.root = self.BDIRS['loc']
dirents.extend(os.listdir(path))
return dirents
# def mythread(self):
#
# """
# The beauty of the FUSE python implementation is that with the python interp
# running in foreground, you can have threads
# """
# print "mythread: started"
# while 1:
# time.sleep(120)
# print "mythread: ticking"
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.
self.assertTrue(today == todayagain or
abs(todayagain - today) < timedelta(seconds=0.5))
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 main():
results = []
log_file = 'database.log'
# with open(log_file) as f:
# #Spawn pool of workers to execute http queries
# pool = Pool()
# results = pool.map_async(run_log, f,1)
# pool.close()
# while not results.ready():
# remaining = results._number_left
# print "Waiting for", remaining, "tasks to complete..."
# sys.stdout.flush()
# time.sleep(10)
with open(log_file) as in_, tqdm(total=40000) as pbar:
count = 0.
last_timestamp = datetime.utcnow()
for l_ in in_:
count += 1
res, last_timestamp = run_log(l_, last_timestamp)
if len(results) > 40000:
break
if count == 19:
count = 0
pbar.update(19)
sys.stdout.flush()
if res != -1.:
results.append(res)
with open(log_file + '-test2', 'a') as out:
for entry in results:
# for entry in results.get():
if entry is not None:
out.write(str(entry))
def run_log(query_line, last_timestamp):
# open queries and regex for links
url_ = re.findall('"GET (.*?) HTTP', query_line)
last_timestamp_new = datetime.utcnow()
if len(url_) == 1:
request_url = url_[0]
query_times = []
resp = ''
result_size = 0
try:
utcnow = datetime.utcnow()
midnight_utc = datetime.combine(utcnow.date(), time(0))
delta_last_query = (datetime.utcnow() - last_timestamp).total_seconds()
for _ in range(11):
response, exec_time = run_http_request(request_url)
# if exec_time == -1.:
# break
query_times.append(exec_time)
# timesleep.sleep(random.random()*0.1)
last_timestamp_new = datetime.utcnow()
timestamp_query = ((last_timestamp_new - midnight_utc).total_seconds())
respJson = response.json()
result_size = get_result_size(respJson)
except:
exec_time = -1
if exec_time != -1 and len(query_times) == 11: #and result_size > 0:
cold_exec_time = query_times[0]
warm_times = query_times[1:]
warm_mean = np.mean(warm_times, dtype=np.float64)
time_vec = [timestamp_query, delta_last_query]
query_clean = cleanup_query(request_url)
res = str(query_clean + '\t'+ str(time_vec) + '\t' + str(warm_mean) + '\t' + str(cold_exec_time) + '\t' + str(result_size) + '\n')
return (res, last_timestamp_new)
else:
return (-1., last_timestamp_new)
else:
return (-1., last_timestamp_new)
