Modified the make_lulc_rasters function to make_simple_raster so that it only makes one raster at a time.
Replaced args with the file path to reduce side effects of potentially modifying args. Adjusted code styles based on SublimeLinter. See #BITBUCKET-3758.
This commit is contained in:
Joanna Lin
parent
6db9d3966f
commit
875366ee5a
|
|
@ -10,85 +10,84 @@ import numpy
|
|||
import pygeoprocessing.testing
|
||||
|
||||
|
||||
def make_lulc_rasters(args, raster_keys, start_val):
|
||||
"""Create LULC rasters with specified raster names and starting value.
|
||||
def make_simple_raster(raster_path, fill_val):
|
||||
"""Create a 10x10 raster with designated path and fill value.
|
||||
|
||||
Parameters:
|
||||
args (dict): the arguments used in the testing function.
|
||||
raster_keys (list): a list of raster name(s) that are either
|
||||
'lulc_cur_path', 'lulc_fut_path' or 'lulc_redd_path'.
|
||||
start_val (int): the starting value used for filling the rasters(s).
|
||||
raster_path (str): a raster path for making the new raster.
|
||||
fill_val (int): the value used for filling the raster.
|
||||
|
||||
Returns:
|
||||
None.
|
||||
lulc_path (str): the path of the raster file.
|
||||
|
||||
"""
|
||||
srs = osr.SpatialReference()
|
||||
srs.ImportFromEPSG(26910)
|
||||
srs.ImportFromEPSG(26910) # UTM Zone 10N
|
||||
projection_wkt = srs.ExportToWkt()
|
||||
|
||||
for val, key in enumerate(raster_keys, start=start_val):
|
||||
lulc_array = numpy.empty((10, 10))
|
||||
lulc_array.fill(val)
|
||||
lulc_path = os.path.join('.', key+'.tif')
|
||||
pygeoprocessing.testing.create_raster_on_disk(
|
||||
[lulc_array], (461261, 4923265), projection_wkt, -1, (1, -1),
|
||||
filename=lulc_path)
|
||||
args[key] = lulc_path
|
||||
lulc_array = numpy.empty((10, 10))
|
||||
lulc_array.fill(fill_val)
|
||||
pygeoprocessing.testing.create_raster_on_disk(
|
||||
[lulc_array],
|
||||
(461261, 4923265), # Origin based on the projection
|
||||
projection_wkt,
|
||||
-1,
|
||||
(1, -1),
|
||||
filename=raster_path)
|
||||
|
||||
|
||||
def assert_npv(args, actual_npv, out_npv_filename):
|
||||
"""Assert that the output npv array is the same as the npv array
|
||||
computed manually based on the synthetic data.
|
||||
|
||||
def assert_npv(out_npv_filename, actual_npv):
|
||||
"""Assert that the output npv is the same as the actual npv.
|
||||
|
||||
Parameters:
|
||||
args (dict): the arguments used in the testing function.
|
||||
out_npv_filename (str): the filename of the output npv TIFF file.
|
||||
actual_npv (float): the actual npv to be filled in the array.
|
||||
out_npv_filename (string): the filename of the output npv TIFF file.
|
||||
|
||||
Returns:
|
||||
None.
|
||||
"""
|
||||
actual_npv_arr = numpy.empty((10, 10))
|
||||
actual_npv_arr.fill(actual_npv)
|
||||
|
||||
out_npv_raster = gdal.OpenEx(os.path.join(args['workspace_dir'], out_npv_filename))
|
||||
"""
|
||||
out_npv_raster = gdal.OpenEx(out_npv_filename, gdal.OF_RASTER)
|
||||
out_npv_raster_band = out_npv_raster.GetRasterBand(1)
|
||||
out_npv_arr = out_npv_raster_band.ReadAsArray()
|
||||
|
||||
actual_npv_arr = numpy.empty((10, 10))
|
||||
actual_npv_arr.fill(actual_npv)
|
||||
numpy.testing.assert_almost_equal(actual_npv_arr, out_npv_arr)
|
||||
|
||||
|
||||
def make_pools_csv(args):
|
||||
def make_pools_csv(pools_csv_path):
|
||||
"""Create a carbon pools csv file with simplified land cover types.
|
||||
|
||||
Parameters:
|
||||
args (dict): the arguments used in the testing function.
|
||||
pools_csv_path (str): the path of carbon pool csv.
|
||||
|
||||
Returns:
|
||||
None.
|
||||
|
||||
"""
|
||||
pools_csv = os.path.join(args['workspace_dir'], 'pools.csv')
|
||||
with open(pools_csv, 'w') as open_table:
|
||||
with open(pools_csv_path, 'w') as open_table:
|
||||
open_table.write('C_above,C_below,C_soil,C_dead,lucode,LULC_Name\n')
|
||||
open_table.write('15,10,60,1,1,"lulc code 1"\n')
|
||||
open_table.write('5,3,20,0,2,"lulc code 2"\n')
|
||||
open_table.write('2,1,5,0,3,"lulc code 3"\n')
|
||||
args['carbon_pools_path'] = pools_csv
|
||||
|
||||
|
||||
class CarbonTests(unittest.TestCase):
|
||||
"""Tests for the Carbon Model."""
|
||||
|
||||
def setUp(self):
|
||||
"""Overriding setUp function to create temp workspace directory."""
|
||||
"""Override setUp function to create temp workspace directory."""
|
||||
# this lets us delete the workspace after its done no matter the
|
||||
# the rest result
|
||||
self.workspace_dir = tempfile.mkdtemp()
|
||||
|
||||
def tearDown(self):
|
||||
"""Overriding tearDown function to remove temporary directory."""
|
||||
"""Override tearDown function to remove temporary directory."""
|
||||
shutil.rmtree(self.workspace_dir)
|
||||
|
||||
|
||||
def test_carbon_full_fast(self):
|
||||
"""Carbon: full model run with synthetic data."""
|
||||
def test_carbon_full(self):
|
||||
"""Carbon: full model run."""
|
||||
from natcap.invest import carbon
|
||||
|
||||
args = {
|
||||
|
|
@ -101,17 +100,27 @@ class CarbonTests(unittest.TestCase):
|
|||
u'discount_rate': -7.1,
|
||||
}
|
||||
|
||||
make_lulc_rasters(args, ['lulc_cur_path', 'lulc_fut_path', 'lulc_redd_path'], 1)
|
||||
make_pools_csv(args)
|
||||
# Create LULC rasters and pools csv in workspace and add them to args.
|
||||
lulc_names = ['lulc_cur_path', 'lulc_fut_path', 'lulc_redd_path']
|
||||
for fill_val, lulc_name in enumerate(lulc_names, 1):
|
||||
args[lulc_name] = os.path.join(args['workspace_dir'],
|
||||
lulc_name + '.tif')
|
||||
make_simple_raster(args[lulc_name], fill_val)
|
||||
|
||||
args['carbon_pools_path'] = os.path.join(args['workspace_dir'],
|
||||
'pools.csv')
|
||||
make_pools_csv(args['carbon_pools_path'])
|
||||
|
||||
carbon.execute(args)
|
||||
|
||||
#Add assertions for npv for future and REDD scenarios
|
||||
assert_npv(args, -0.34220789207450352, 'npv_fut.tif')
|
||||
assert_npv(args, -0.4602106134795047, 'npv_redd.tif')
|
||||
# Add assertions for npv for future and REDD scenarios
|
||||
assert_npv(os.path.join(args['workspace_dir'], 'npv_fut.tif'),
|
||||
-0.3422078)
|
||||
assert_npv(os.path.join(args['workspace_dir'], 'npv_redd.tif'),
|
||||
-0.4602106)
|
||||
|
||||
|
||||
def test_carbon_future_fast(self):
|
||||
"""Carbon: regression testing future scenario using synthetic data."""
|
||||
def test_carbon_future(self):
|
||||
"""Carbon: regression testing future scenario."""
|
||||
from natcap.invest import carbon
|
||||
args = {
|
||||
u'workspace_dir': self.workspace_dir,
|
||||
|
|
@ -123,22 +132,39 @@ class CarbonTests(unittest.TestCase):
|
|||
u'discount_rate': -7.1,
|
||||
}
|
||||
|
||||
make_lulc_rasters(args, ['lulc_cur_path', 'lulc_fut_path'], 1)
|
||||
make_pools_csv(args)
|
||||
lulc_names = ['lulc_cur_path', 'lulc_fut_path']
|
||||
for fill_val, lulc_name in enumerate(lulc_names, 1):
|
||||
args[lulc_name] = os.path.join(args['workspace_dir'],
|
||||
lulc_name + '.tif')
|
||||
make_simple_raster(args[lulc_name], fill_val)
|
||||
|
||||
args['carbon_pools_path'] = os.path.join(args['workspace_dir'],
|
||||
'pools.csv')
|
||||
make_pools_csv(args['carbon_pools_path'])
|
||||
|
||||
carbon.execute(args)
|
||||
#Add assertions for npv for the future scenario
|
||||
assert_npv(args, -0.34220789207450352, 'npv_fut.tif')
|
||||
# Add assertions for npv for the future scenario
|
||||
assert_npv(os.path.join(args['workspace_dir'], 'npv_fut.tif'),
|
||||
-0.3422078)
|
||||
|
||||
|
||||
def test_carbon_missing_landcover_values_fast(self):
|
||||
"""Carbon: testing expected exception on incomplete with synthetic data."""
|
||||
def test_carbon_missing_landcover_values(self):
|
||||
"""Carbon: testing expected exception on missing LULC codes."""
|
||||
from natcap.invest import carbon
|
||||
args = {
|
||||
u'workspace_dir': self.workspace_dir,
|
||||
u'do_valuation': False,
|
||||
}
|
||||
|
||||
make_lulc_rasters(args, ['lulc_cur_path', 'lulc_fut_path'], 200)
|
||||
make_pools_csv(args)
|
||||
lulc_names = ['lulc_cur_path', 'lulc_fut_path']
|
||||
for fill_val, lulc_name in enumerate(lulc_names, 200):
|
||||
args[lulc_name] = os.path.join(args['workspace_dir'],
|
||||
lulc_name + '.tif')
|
||||
make_simple_raster(args[lulc_name], fill_val)
|
||||
|
||||
args['carbon_pools_path'] = os.path.join(args['workspace_dir'],
|
||||
'pools.csv')
|
||||
make_pools_csv(args['carbon_pools_path'])
|
||||
|
||||
# Value error should be raised with lulc code 200
|
||||
with self.assertRaises(ValueError):
|
||||
carbon.execute(args)
|
||||
|
|
|
|||
Loading…
Reference in New Issue