-
Notifications
You must be signed in to change notification settings - Fork 0
/
csv_handler.py
101 lines (89 loc) · 3.5 KB
/
csv_handler.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
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
import pandas as pd
import os
import shutil
from typing import Optional
CACHE = "./results_cache/"
INFRASTRUCTURE = 'infrastructure.csv'
APPLICATION = 'application.csv'
FILE = 'run.py'
STATIC = "./static/"
#copy results into cache file
def into_cache(UploadedFile):
df = pd.read_csv(UploadedFile,index_col=0)
temp = pd.read_csv(STATIC + 'temp.csv', index_col = 0)
temp = pd.merge(temp, df, on = 'time',how = 'outer', sort = True)
temp.fillna(method = 'ffill', inplace=True ,axis=0)
temp.to_csv(CACHE + UploadedFile.name)
#clean cache file before new simulation
def clean_cache(path):
if not os.path.exists(path):
os.makedirs(path)
else:
shutil.rmtree(path)
os.mkdir(path)
#check if inf and app exists to modify spaceholder
def check_exists(inf,app):
filelist = os.listdir(CACHE)
if inf and app in filelist:
return 2
elif inf in filelist:
return 1
else:
return False
#read first line to be a base of addrows
def read_first_line(csv):
df = pd.read_csv(CACHE + csv,nrows = 1,index_col=0)
return df
def get_row_length(csv):
df = pd.read_csv(CACHE + csv,index_col=0)
return df.shape[0]
#optimize performance
def read_row_by_sequence(csv,nrows):
df = pd.read_csv(CACHE + csv,nrows=nrows,index_col=0)
return df
def merge_results():
files = os.listdir(CACHE)
temp_inf = pd.read_csv(STATIC+'temp.csv',index_col=0)
temp_app = pd.read_csv(STATIC+'temp.csv',index_col=0)
for f in files:
if '1_' in f and f.endswith('.csv'):
df = pd.read_csv(CACHE + f, index_col=0)
temp_inf = pd.merge(temp_inf,df,on='time',how='outer', sort=True)
temp_inf.fillna(method = 'ffill',inplace=True, axis = 0)
temp_inf.to_csv(CACHE + 'infrastructure.csv')
elif '2_' in f and f.endswith('.csv'):
df = pd.read_csv(CACHE + f, index_col=0)
temp_app = pd.merge(temp_app,df,on='time',how='outer', sort=True)
temp_app.fillna(method = 'ffill',inplace=True, axis = 0)
temp_app.to_csv(CACHE + 'application.csv')
#core function to make the base of lines then front-end can draw it.
def output_csv(PM, rename, filter:Optional[str] = 'sum',type: Optional[int] = 1, delay: Optional[float] = 0):
result_dir = f"results_cache/"
sum_contect = "time,"+ rename +"\n"
dynamic_content = "time,"+ rename+' dynamic' +"\n"
static_content = "time,"+ rename+' static' +"\n"
full_content = "time,"+ rename+' static,' + rename+' dynamic'+"\n"
os.makedirs(result_dir, exist_ok=True)
for i, powermeter in enumerate(PM.measurements):
sum = powermeter.static + powermeter.dynamic
j = i
j *= PM.measurement_interval
if filter == 'sum':
sum_contect += f"{j + delay},{sum}\n"
elif filter == 'dynamic':
dynamic_content += f"{j + delay},{powermeter.dynamic}\n"
elif filter == 'static':
static_content += f"{j + delay},{powermeter.static}\n"
elif filter == 'all':
full_content += f"{j + delay},{powermeter.static},{powermeter.dynamic}\n"
with open(f"{result_dir}/{type}_{rename}.csv", 'w') as csvfile:
if filter == 'sum':
csvfile.write(sum_contect)
elif filter == 'dynamic':
csvfile.write(dynamic_content)
elif filter == 'static':
csvfile.write(static_content)
elif filter == 'all':
csvfile.write(full_content)
def from_filechooser_to_cache(file):
shutil.move(file,CACHE+file)