from copy import deepcopy
import json
import glob
import numpy as np
import os
import re
import shutil
import unittest
from difflib import Differ
from numpy.testing import assert_array_almost_equal
from prms_python import (
modify_params, Parameters, Scenario, ScenarioSeries, Simulation,
SimulationSeries, Data
)
OPJ = os.path.join
class TestSimulationSeries(unittest.TestCase):
'''
Expect that a SimulationSeries
'''
def setUp(self):
self.test_model_data_dir = os.path.join(
'test', 'data', 'models', 'lbcd'
)
self.simulation_dir = os.path.join(self.test_model_data_dir, 'tmp_sim')
def tearDown(self):
for g in glob.glob(OPJ(self.simulation_dir, '*')):
shutil.rmtree(g)
def test_simulation_series(self):
tdd = self.test_model_data_dir
data = Data(OPJ(tdd, 'data'))
base_parameters = Parameters(OPJ(tdd, 'parameters'))
def _copy_mod(base_parameters, val):
ret = deepcopy(base_parameters)
ret['dday_intcp'][:] = val
return ret
parameters_gen = (
(val, _copy_mod(base_parameters, val))
for val in (-50, -40, -30)
)
control_path = OPJ(tdd, 'control')
series = SimulationSeries(
Simulation.from_data(
data, parameters, control_path,
self.simulation_dir + str(val)
)
for val, parameters in parameters_gen
)
outputs = list(series.run().outputs_iter())
self.assertEqual(len(series), 3)
self.assertEqual(len(outputs), 3)
for out in outputs:
sdir = out['simulation_dir']
assert os.path.isdir(out['simulation_dir'])
assert os.path.exists(OPJ(sdir, 'outputs', 'statvar.dat'))
assert os.path.exists(OPJ(sdir, 'inputs', 'data'))
assert os.path.exists(OPJ(sdir, 'inputs', 'parameters'))
shutil.rmtree(sdir)
class TestSimulation(unittest.TestCase):
"""
Simulations should take a base directory and return a simulation directory
"""
def setUp(self):
self.test_data_dir = os.path.join('test', 'data')
self.test_model_data_dir = os.path.join(
'test', 'data', 'models', 'lbcd'
)
self.simulation_dir = os.path.join(self.test_data_dir, 'tmp_sim')
def tearDown(self):
if os.path.exists(self.simulation_dir):
shutil.rmtree(self.simulation_dir)
def test_simulation_no_simdir(self):
"Simulation should run and write outputs to input directory when simulation_dir is not specified"
s = Simulation(self.test_model_data_dir)
s.run()
g = [
os.path.basename(f)
for f in glob.glob(os.path.join(self.test_model_data_dir, '*'))
]
self.assertIn('prms_ic.out', g)
self.assertIn('prms.out', g)
self.assertIn('statvar.dat', g)
self.assertIn('animation.out.nhru', g)
def test_simulation_w_simdir(self):
"Simulation should create sim dir with inputs and outputs directory when simulation_dir is specified"
s = Simulation(self.test_model_data_dir, self.simulation_dir)
s.run()
gs = [
os.path.basename(f)
for f in glob.glob(os.path.join(self.simulation_dir, '*'))
]
self.assertIn('inputs', gs)
self.assertIn('outputs', gs)
assert_valid_input_dir(
self, os.path.join(self.simulation_dir, 'inputs')
)
assert_valid_output_dir(
self, os.path.join(self.simulation_dir, 'outputs')
)
def test_simulation_from_data(self):
"""
Use @classmethod from_data to build a simulation from Parameters and Data instances
"""
tdd = self.test_model_data_dir
data = Data(OPJ(tdd, 'data'))
parameters = Parameters(OPJ(tdd, 'parameters'))
ctrl = OPJ(tdd, 'control')
test_dir = OPJ(tdd, 'test-sim-dir')
if os.path.isdir(test_dir):
shutil.rmtree(test_dir)
s = Simulation.from_data(data, parameters, ctrl, test_dir)
s.run()
g = [
os.path.basename(f)
for f in glob.glob(OPJ(test_dir, 'outputs', '*'))
]
self.assertIn('prms_ic.out', g)
self.assertIn('prms.out', g)
self.assertIn('statvar.dat', g)
self.assertIn('animation.out.nhru', g)
# clean up
shutil.rmtree(test_dir)
class TestScenario(unittest.TestCase):
def setUp(self):
self.test_data_dir = os.path.join('test', 'data')
self.test_model_data_dir = os.path.join(
'test', 'data', 'models', 'lbcd'
)
self.scenario_dir = os.path.join(self.test_data_dir, 'tmp_scenario')
def tearDown(self):
if os.path.exists(self.scenario_dir):
shutil.rmtree(self.scenario_dir)
def test_create_scenario(self):
"""a simulation setup should create a simulation directory with correct scenario data"""
s = Scenario(
self.test_model_data_dir, self.scenario_dir,
title='Scenario Uno', description='test scenario for prms_python'
)
param_mods = {
'snow_adj': lambda x: 1.1*x,
'rad_trncf': lambda x: 0.9*x
}
s.build(param_mod_funs=param_mods)
assert_valid_input_dir(self, self.scenario_dir) # os.path.join(self.scenario_dir, 'inputs'))
s.run()
assert_valid_input_dir(
self, os.path.join(self.scenario_dir, 'inputs')
)
assert_valid_output_dir(
self, os.path.join(self.scenario_dir, 'outputs')
)
md_json_path = os.path.join(self.scenario_dir, 'metadata.json')
assert os.path.isfile(md_json_path)
md_json = json.loads(open(md_json_path).read())
assert md_json['title'] == 'Scenario Uno'
assert md_json['description'] == 'test scenario for prms_python'
assert 'start_datetime' in md_json
assert 'end_datetime' in md_json
assert 'mod_funs_dict' in md_json
p_base = Parameters(
os.path.join(self.test_model_data_dir, 'parameters')
)
p_scen = Parameters(
os.path.join(self.scenario_dir, 'inputs', 'parameters')
)
assert_array_almost_equal(p_base['snow_adj']*1.1, p_scen['snow_adj'])
assert_array_almost_equal(p_base['rad_trncf']*0.9, p_scen['rad_trncf'])
class TestScenarios(unittest.TestCase):
def setUp(self):
self.test_data_dir = os.path.join('test', 'data')
self.test_model_data_dir = os.path.join(
self.test_data_dir, 'models', 'lbcd'
)
self.scenarios_dir = os.path.join(self.test_data_dir, 'tmp_scenarios')
def tearDown(self):
if os.path.exists(self.scenarios_dir):
shutil.rmtree(self.scenarios_dir)
def test_scenario_series(self):
"create_many_simulations should create many simulation directories and correct data"
s = ScenarioSeries(
self.test_model_data_dir, self.scenarios_dir,
title='scenario series uno',
description='''
Each scenario is given a title with the schema
'"