-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtime_counter.py
51 lines (40 loc) · 1.67 KB
/
time_counter.py
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
import threading
import time
import functools
from conversation_history import conversation_history as ch
# decorator is a cool way of timing the ongoing task, however you can't read global variables in it
# is there any other way of adding that time to time_history? i suppose not
def timeit(func):
@functools.wraps(func)
def newfunc(*args, **kwargs):
startTime = time.time()
func(*args, **kwargs)
elapsedTime = time.time() - startTime
formatted_time = int(elapsedTime*1000)/1000
print('function [{}] finished in {} s'.format(
func.__name__, formatted_time))
return newfunc
# global conversation history object
conversationHistoryController = ch.ConversationHistory()
def sample_conversation(x):
startTime = time.time()
conversation = conversationHistoryController.newConversation()
for i in range(x):
print(i)
elapsedTime = time.time() - startTime
formatted_time = int(elapsedTime * 1000) / 1000
conversation["duration"] = formatted_time
conversationHistoryController.addConversation(conversation)
def background_ongoing_task(time_to_sleep):
while(True):
time.sleep(time_to_sleep)
conversationHistoryController.readConversationHistory()
if __name__ == "__main__":
background_task = threading.Thread(target=background_ongoing_task, args=(3,))
background_task.start()
threadVar = threading.Thread(target=sample_conversation, args=(100000,))
threadVar.start()
threadVar_1 = threading.Thread(target=sample_conversation, args=(200000,))
threadVar_1.start()
threadVar_2 = threading.Thread(target=sample_conversation, args=(300000,))
threadVar_2.start()