-
Notifications
You must be signed in to change notification settings - Fork 0
/
data_scripts.py
169 lines (126 loc) · 5.4 KB
/
data_scripts.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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
import os
import re
from pathlib import Path
from typing import Sequence
import numpy as np
from solver_selector.data_structures import (
NonlinearSolverStats,
PerformancePredictionData,
SolverSelectionData,
)
def make_solution_stats(
perf: Sequence[SolverSelectionData], converged=True
) -> Sequence[NonlinearSolverStats]:
solution_stats = np.array([data.nonlinear_solver_stats for data in perf])
if not converged:
return solution_stats
converged_indices = make_converged_indices(perf)
return solution_stats[converged_indices]
def make_converged_indices(perf: Sequence[SolverSelectionData]):
solution_stats = make_solution_stats(perf, converged=False)
return np.array([x.is_converged for x in solution_stats])
def make_solve_linear_system_time(perf, converged=True):
solution_stats_converged = make_solution_stats(perf, converged=converged)
return np.array(
[
y.solve_linear_system_time
for x in solution_stats_converged
for y in x.iterations
]
)
def make_assemble_time(perf, converged=True):
solution_stats_converged = make_solution_stats(perf, converged=converged)
return np.array(
[y.assembly_time for x in solution_stats_converged for y in x.iterations]
)
def make_peclet_max(perf, converged=True):
predictions = make_predictions(perf, converged=converged)
return np.array([x.context.peclet_max for x in predictions])
def make_peclet_mean(perf, converged=True):
predictions_converged = make_predictions(perf, converged=converged)
return np.array([x.context.peclet_mean for x in predictions_converged])
def make_cfl_max(perf):
predictions = make_predictions(perf)
return np.array([x.context.cfl_max for x in predictions])
def make_cfl_mean(perf):
predictions = make_predictions(perf)
return np.array([x.context.cfl_mean for x in predictions])
def make_time_step(perf, converged=True):
predictions = make_predictions(perf, converged=converged)
return np.array([x.context.time_step_value for x in predictions])
def make_inlet_rate(perf, converged=True):
predictions = make_predictions(perf, converged=converged)
array = np.array([x.context.inlet_rate for x in predictions])
return np.maximum(array, 1e-6)
def make_outlet_rate(perf, converged=True):
predictions = make_predictions(perf, converged=converged)
array = np.array([x.context.outlet_rate for x in predictions])
return np.maximum(array, 1e-6)
def make_simulation_time(perf, converged=True):
time_steps = make_time_step(perf, converged=converged)
if not converged:
conv = make_converged_indices(perf)
time_steps[~conv] = 0
return np.cumsum(time_steps)
def make_predictions(
perf: Sequence[SolverSelectionData], converged=True
) -> Sequence[PerformancePredictionData]:
predictions = np.array([data.prediction for data in perf])
if converged:
converged_indices = make_converged_indices(perf)
predictions = predictions[converged_indices]
return predictions
def sum_per_time_step(values, perf, converged=True):
time_step_indices = make_time_step_numbers(perf, converged=converged)
res = np.zeros(time_step_indices[-1] + 1)
for i, val in enumerate(values):
res[time_step_indices[i]] += val
return res
def make_time_step_numbers(perf, converged=True):
solution_stats = make_solution_stats(perf, converged=converged)
return np.array([i for i, x in enumerate(solution_stats) for y in x.iterations])
def make_num_nonlinear_iters(perf, converged=True):
solution_stats_converged = make_solution_stats(perf, converged=converged)
return [len(x.iterations) for x in solution_stats_converged]
def make_num_linear_iters(perf, converged=True):
solution_stats_converged = make_solution_stats(perf, converged=converged)
return np.array(
[
y.num_linear_iterations
for x in solution_stats_converged
for y in x.iterations
]
)
def append_experiment_name(path: str | Path) -> Path:
path = Path(path).absolute()
filename = path.name
experiment_dir = path.parent / "performance"
experiment_dir.mkdir(exist_ok=True)
name = filename.removesuffix(".npy").removesuffix(".py")
ids = ["-1"]
for file in os.listdir(experiment_dir):
if file.startswith(name):
match = re.findall(r"(\d+).npy", file)
ids.extend(match)
int_ids = [int(x) for x in ids]
max_id = max(int_ids) + 1
experiment_dir.mkdir(exist_ok=True)
return experiment_dir / f"{name}_{max_id}.npy"
def get_newest_data_paths(experiment_name: str, n_newest=3) -> Sequence[str]:
name = Path(experiment_name).name.removesuffix(".py").removesuffix(".npy")
work_dir = Path(experiment_name).absolute().parent / "performance"
data = {}
for fname in os.listdir(work_dir):
match = re.findall(rf"{name}_(\d+).npy", fname)
if len(match) > 0:
data[int(match[0])] = fname
keys = sorted(data)[-n_newest:]
values = [str(work_dir / data[key]) for key in keys]
return values
def load_data(experiment_name, n_newest) -> Sequence[Sequence[SolverSelectionData]]:
paths = get_newest_data_paths(experiment_name, n_newest=n_newest)
assert len(paths) == n_newest, "Data not found"
print("Loading data:")
for path in paths:
print(path)
return [np.load(path, allow_pickle=True).tolist() for path in paths]