-
Notifications
You must be signed in to change notification settings - Fork 14
/
utils.py
96 lines (74 loc) · 2.72 KB
/
utils.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
import json
import os
import subprocess
from tempfile import TemporaryDirectory
import numpy as np
import pandas as pd
# global boolean to perform coverage testing
# (run executables instead of Python interface, much slower)
from pytest import coverage
import pysvzerod
this_file_dir = os.path.abspath(os.path.dirname(__file__))
RTOL_PRES = 1.0e-7
RTOL_FLOW = 1.0e-8
def execute_pysvzerod(testfile, mode):
"""Execute pysvzerod (via Python interface or executable).
Args:
testfile: Path to the input file.
mode: svZeroDSolver application (solver or calibrator).
"""
assert mode in ["solver", "calibrator"], "unknown mode: " + mode
# read configuration
with open(testfile) as ff:
config = json.load(ff)
if coverage:
# run via executable (slow)
with TemporaryDirectory() as tempdir:
out_name = os.path.join(tempdir, "out")
exe = os.path.join(this_file_dir, "..", "Release", "svzerod")
subprocess.run([exe + mode, testfile, out_name])
if mode == "solver":
result = pd.read_csv(out_name)
elif mode == "calibrator":
with open(out_name) as ff:
result = json.load(ff)
else:
# run via Python binding (fast)
if mode == "solver":
result = pysvzerod.simulate(config)
elif mode == "calibrator":
result = pysvzerod.calibrate(config)
return result, config
def run_test_case_by_name(name, output_variable_based=False, folder="."):
"""Run a test case by its case name.
Args:
name: Name of the test case.
testdir: Directory for performing the simulation.
"""
# file name of test case
testfile = os.path.join(this_file_dir, "cases", name + ".json")
# run test
result, config = execute_pysvzerod(testfile, "solver")
if not output_variable_based:
output = {
"pressure_in": {},
"pressure_out": {},
"flow_in": {},
"flow_out": {},
}
for vessel in config["vessels"]:
name = vessel["vessel_name"]
branch_id, seg_id = name.split("_")
branch_id, seg_id = int(branch_id[6:]), int(seg_id[3:])
for f in ["pressure", "flow"]:
for l in ["in"] * (seg_id == 0) + ["out"]:
n = f + "_" + l
ids = result.name == name
output[n][branch_id] = np.array(result[ids][n])
else:
output = result
return output
def get_result(result_array, field, branch, time_step):
""" "Get results at specific field, branch, branch_node and time step."""
# extract result
return result_array[field][branch][time_step]