Newer
Older
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
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
'''
Created on 05/11/2015
@author: MMPE
'''
from __future__ import unicode_literals
from __future__ import print_function
from __future__ import division
from __future__ import absolute_import
from future import standard_library
standard_library.install_aliases()
import unittest
import os
import filecmp
import shutil
from zipfile import ZipFile
import numpy as np
import pandas as pd
from wetb.prepost import dlctemplate as tmpl
from wetb.prepost import Simulations as sim
from wetb.prepost.misc import DictDiff
class Template(unittest.TestCase):
def setUp(self):
self.basepath = os.path.dirname(__file__)
class TestErrorLogs(Template):
def test_loganalysis(self):
# select a few log cases and do the analysis, save to csv and convert
# saved result to DataFrame
pass
class TestGenerateInputs(Template):
def test_launch_dlcs_excel(self):
# manually configure paths, HAWC2 model root path is then constructed as
# p_root_remote/PROJECT/sim_id, and p_root_local/PROJECT/sim_id
# adopt accordingly when you have configured your directories differently
p_root = os.path.join(self.basepath, 'data/')
# project name, sim_id, master file name
tmpl.PROJECT = 'demo_dlc'
tmpl.MASTERFILE = 'demo_dlc_master_A0001.htc'
# MODEL SOURCES, exchanche file sources
tmpl.P_RUN = os.path.join(p_root, tmpl.PROJECT, 'remote/')
tmpl.P_SOURCE = os.path.join(p_root, tmpl.PROJECT, 'source/')
# location of the master file
tmpl.P_MASTERFILE = os.path.join(p_root, tmpl.PROJECT,
'source', 'htc', '_master/')
# location of the pre and post processing data
tmpl.POST_DIR = os.path.join(p_root, tmpl.PROJECT, 'remote',
'prepost/')
# make sure the remote dir is empty so a test does not pass on data
# generated during a previous cycle
if os.path.exists(os.path.join(p_root, tmpl.PROJECT, 'remote')):
shutil.rmtree(os.path.join(p_root, tmpl.PROJECT, 'remote'))
tmpl.force_dir = tmpl.P_RUN
tmpl.launch_dlcs_excel('remote', silent=True, runmethod='gorm',
pbs_turb=True, zipchunks=True,
postpro_node_zipchunks=False,
postpro_node=False)
# we can not check-in empty dirs so we can not compare the complete
# directory structure withouth manually creating the empty dirs here
for subdir in ['control', 'data', 'htc', 'pbs_in', 'pbs_in_turb',
'htc/_master', 'htc/dlc01_demos', 'pbs_in/dlc01_demos',
'zip-chunks-gorm', 'zip-chunks-jess']:
remote = os.path.join(p_root, tmpl.PROJECT, 'remote', subdir)
ref = os.path.join(p_root, tmpl.PROJECT, 'ref', subdir)
# the zipfiles are taken care of separately
ignore = ['remote_chnk_00000.zip']
cmp = filecmp.dircmp(remote, ref, ignore=ignore)
self.assertEqual(len(cmp.diff_files), 0,
"{} {}".format(subdir, cmp.diff_files))
self.assertEqual(len(cmp.right_only), 0,
"{} {}".format(subdir, cmp.right_only))
self.assertEqual(len(cmp.left_only), 0,
"{} {}".format(subdir, cmp.left_only))
# compare the zip files
for fname in ['demo_dlc_remote.zip',
'zip-chunks-gorm/remote_chnk_00000.zip',
'zip-chunks-jess/remote_chnk_00000.zip']:
remote = os.path.join(p_root, tmpl.PROJECT, 'remote', fname)
ref = os.path.join(p_root, tmpl.PROJECT, 'ref', fname)
with ZipFile(remote) as zrem, ZipFile(ref) as zref:
self.assertEqual(len(zrem.infolist()), len(zref.infolist()))
frem = {f.filename:f.file_size for f in zrem.infolist()}
fref = {f.filename:f.file_size for f in zref.infolist()}
dd = DictDiff(frem, fref)
self.assertEqual(len(dd.added()), 0,
"{} {}".format(fname, dd.added()))
self.assertEqual(len(dd.removed()), 0,
"{} {}".format(fname, dd.removed()))
self.assertEqual(len(dd.changed()), 0,
"{} {}".format(fname, dd.changed()))
# for the pickled file we can just read it
remote = os.path.join(p_root, tmpl.PROJECT, 'remote', 'prepost')
ref = os.path.join(p_root, tmpl.PROJECT, 'ref', 'prepost')
cmp = filecmp.cmp(os.path.join(remote, 'remote_tags.txt'),
os.path.join(ref, 'remote_tags.txt'), shallow=False)
self.assertTrue(cmp)
# with open(os.path.join(remote, 'remote.pkl'), 'rb') as FILE:
# pkl_remote = pickle.load(FILE)
# with open(os.path.join(ref, 'remote.pkl'), 'rb') as FILE:
# pkl_ref = pickle.load(FILE)
# self.assertTrue(pkl_remote == pkl_ref)
class TestFatigueLifetime(Template):
def test_leq_life(self):
"""Verify if prepost.Simulation.Cases.fatigue_lifetime() returns
the expected life time equivalent load.
"""
# ---------------------------------------------------------------------
# very simple case
cases = {'case1':{'[post_dir]':'no-path', '[sim_id]':'A0'},
'case2':{'[post_dir]':'no-path', '[sim_id]':'A0'}}
cc = sim.Cases(cases)
fh_list = [('case1', 10/3600), ('case2', 20/3600)]
dfs = pd.DataFrame({'m=1.0' : [2, 3],
'channel' : ['channel1', 'channel1'],
'[case_id]' : ['case1', 'case2']})
neq_life = 1.0
df_Leq = cc.fatigue_lifetime(dfs, neq_life, fh_lst=fh_list,
save=False, update=False, csv=False,
xlsx=False, silent=False)
np.testing.assert_allclose(df_Leq['m=1.0'].values, 2*10 + 3*20)
self.assertTrue(df_Leq['channel'].values[0]=='channel1')
# ---------------------------------------------------------------------
# slightly more complicated
neq_life = 3.0
df_Leq = cc.fatigue_lifetime(dfs, neq_life, fh_lst=fh_list,
save=False, update=False, csv=False,
xlsx=False, silent=False)
np.testing.assert_allclose(df_Leq['m=1.0'].values,
(2*10 + 3*20)/neq_life)
# ---------------------------------------------------------------------
# a bit more complex and also test the sorting of fh_lst and dfs
cases = {'case1':{'[post_dir]':'no-path', '[sim_id]':'A0'},
'case2':{'[post_dir]':'no-path', '[sim_id]':'A0'},
'case3':{'[post_dir]':'no-path', '[sim_id]':'A0'},
'case4':{'[post_dir]':'no-path', '[sim_id]':'A0'}}
cc = sim.Cases(cases)
fh_list = [('case3', 10/3600), ('case2', 20/3600),
('case1', 50/3600), ('case4', 40/3600)]
dfs = pd.DataFrame({'m=3.0' : [2, 3, 4, 5],
'channel' : ['channel1']*4,
'[case_id]' : ['case4', 'case2', 'case3', 'case1']})
neq_life = 5.0
df_Leq = cc.fatigue_lifetime(dfs, neq_life, fh_lst=fh_list,
save=False, update=False, csv=False,
xlsx=False, silent=False)
expected = ((2*2*2*40 + 3*3*3*20 + 4*4*4*10 + 5*5*5*50)/5)**(1/3)
np.testing.assert_allclose(df_Leq['m=3.0'].values, expected)
# ---------------------------------------------------------------------
# more cases and with sorting
base = {'[post_dir]':'no-path', '[sim_id]':'A0'}
cases = {'case%i' % k : base for k in range(50)}
cc = sim.Cases(cases)
# reverse the order of how they appear in dfs and fh_lst
fh_list = [('case%i' % k, k*10/3600) for k in range(49,-1,-1)]
dfs = pd.DataFrame({'m=5.2' : np.arange(1,51,1),
'channel' : ['channel1']*50,
'[case_id]' : ['case%i' % k for k in range(50)]})
df_Leq = cc.fatigue_lifetime(dfs, neq_life, fh_lst=fh_list,
save=False, update=False, csv=False,
xlsx=False, silent=False)
expected = np.sum(np.power(np.arange(1,51,1), 5.2)*np.arange(0,50,1)*10)
expected = np.power(expected/neq_life, 1/5.2)
np.testing.assert_allclose(df_Leq['m=5.2'].values, expected)
if __name__ == "__main__":
unittest.main()