VeniVidiVici
/
invest
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

reorganization, linting, cleanup

This commit is contained in:
Claire Simpson 2025-02-06 12:38:05 -07:00
parent 068f37c7ce
commit 7e0eb0d0df
5 changed files with 60 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
tests/test_annual_water_yield.py
View File

@ -414,7 +414,6 @@ class AnnualWaterYieldTests(unittest.TestCase):
validation_warnings[0],
(['demand_table_path'], 'Input is required but has no value'))
def test_fractp_op(self):
"""Test `fractp_op`"""
from natcap.invest.annual_water_yield import fractp_op
@ -447,7 +446,7 @@ class AnnualWaterYieldTests(unittest.TestCase):
and `compute_water_yield_volume`"""
from natcap.invest import annual_water_yield
def create_watershed_results_vector(path_to_shp):
def _create_watershed_results_vector(path_to_shp):
"""Generate a fake watershed results vector file."""
shapely_geometry_list = [
Polygon([(0, 0), (1, 0), (1, 1), (0, 1), (0, 0)]),
@ -468,7 +467,7 @@ class AnnualWaterYieldTests(unittest.TestCase):
vector_format, fields,
attribute_list)
def validate_fields(vector_path, field_name, expected_values, error_msg):
def _validate_fields(vector_path, field_name, expected_values, error_msg):
"""
Validate a specific field in the watershed results vector
by comparing actual to expected values. Expected values generated
@ -492,7 +491,7 @@ class AnnualWaterYieldTests(unittest.TestCase):
# generate fake watershed results vector
watershed_results_vector_path = os.path.join(self.workspace_dir,
"watershed_results.shp")
create_watershed_results_vector(watershed_results_vector_path)
_create_watershed_results_vector(watershed_results_vector_path)
# generate fake val_df
val_df = pandas.DataFrame({'efficiency': [.7, .8], 'height': [12, 50],
@ -503,24 +502,23 @@ class AnnualWaterYieldTests(unittest.TestCase):
# test water yield volume
annual_water_yield.compute_water_yield_volume(
watershed_results_vector_path)
validate_fields(watershed_results_vector_path, "wyield_vol",
[990.0, 800.0],
"Error with water yield volume calculation.")
_validate_fields(watershed_results_vector_path, "wyield_vol",
[990.0, 800.0],
"Error with water yield volume calculation.")
# test rsupply volume
annual_water_yield.compute_rsupply_volume(
watershed_results_vector_path)
validate_fields(watershed_results_vector_path, "rsupply_vl",
[940.0, 730.0],
"Error calculating total realized water supply volume.")
_validate_fields(watershed_results_vector_path, "rsupply_vl",
[940.0, 730.0],
"Error calculating total realized water supply volume.")
# test compute watershed valuation
annual_water_yield.compute_watershed_valuation(
watershed_results_vector_path, val_df)
validate_fields(watershed_results_vector_path, "hp_energy",
[19.329408, 55.5968],
"Error calculating energy.")
validate_fields(watershed_results_vector_path, "hp_val",
[501.9029748723, 4587.91946857059],
"Error calculating net present value.")
_validate_fields(watershed_results_vector_path, "hp_energy",
[19.329408, 55.5968],
"Error calculating energy.")
_validate_fields(watershed_results_vector_path, "hp_val",
[501.9029748723, 4587.91946857059],
"Error calculating net present value.")

57
tests/test_carbon.py
View File

@ -51,38 +51,6 @@ def make_simple_raster(base_raster_path, fill_val, nodata_val):
new_raster = None
def make_simple_lulc_raster(base_raster_path):
"""Create a 2x2 raster on designated path with arbitrary lulc codes.
Args:
base_raster_path (str): the raster path for making the new raster.
Returns:
None.
"""
array = numpy.array([[1, 1], [2, 3]], dtype=int)
srs = osr.SpatialReference()
srs.ImportFromEPSG(26910) # UTM Zone 10N
projection_wkt = srs.ExportToWkt()
# origin hand-picked for this epsg:
geotransform = [461261, 1.0, 0.0, 4923265, 0.0, -1.0]
n = 2
gtiff_driver = gdal.GetDriverByName('GTiff')
new_raster = gtiff_driver.Create(
base_raster_path, n, n, 1, gdal.GDT_Int32, options=[
'TILED=YES', 'BIGTIFF=YES', 'COMPRESS=LZW',
'BLOCKXSIZE=16', 'BLOCKYSIZE=16'])
new_raster.SetProjection(projection_wkt)
new_raster.SetGeoTransform(geotransform)
new_band = new_raster.GetRasterBand(1)
new_band.WriteArray(array)
new_raster.FlushCache()
new_band = None
new_raster = None
def assert_raster_equal_value(base_raster_path, val_to_compare):
"""Assert that the entire output raster has the same value as specified.
@ -304,9 +272,32 @@ class CarbonTests(unittest.TestCase):
"""Test `_generate_carbon_map`"""
from natcap.invest.carbon import _generate_carbon_map
def _make_simple_lulc_raster(base_raster_path):
"""Create a raster on designated path with arbitrary values.
Args:
base_raster_path (str): the raster path for making the new raster.
Returns:
None.
"""
array = numpy.array([[1, 1], [2, 3]], dtype=numpy.int32)
# UTM Zone 10N
srs = osr.SpatialReference()
srs.ImportFromEPSG(26910)
projection_wkt = srs.ExportToWkt()
origin = (461251, 4923245)
pixel_size = (1, 1)
no_data = -999
pygeoprocessing.numpy_array_to_raster(
array, no_data, pixel_size, origin, projection_wkt,
base_raster_path)
# generate a fake lulc raster
lulc_path = os.path.join(self.workspace_dir, "lulc.tif")
make_simple_lulc_raster(lulc_path)
_make_simple_lulc_raster(lulc_path)
# make fake carbon pool dict
carbon_pool_by_type = {1: 5000, 2: 60, 3: 120}

44
tests/test_coastal_blue_carbon.py
View File

@ -26,36 +26,24 @@ LOGGER = logging.getLogger(__name__)
def make_raster_from_array(base_raster_path, array):
"""Create a raster on designated path with arbitrary lulc codes.
"""Create a raster on designated path with arbitrary values.
Args:
base_raster_path (str): the raster path for making the new raster.
array (array): array to save as raster
Returns:
None.
"""
# UTM Zone 10N
srs = osr.SpatialReference()
srs.ImportFromEPSG(26910) # UTM Zone 10N
srs.ImportFromEPSG(26910)
projection_wkt = srs.ExportToWkt()
# origin hand-picked for this epsg:
geotransform = [461261, 1.0, 0.0, 4923265, 0.0, -1.0]
gtiff_driver = gdal.GetDriverByName('GTiff')
new_raster = gtiff_driver.Create(
base_raster_path, array.shape[0], array.shape[1], 1,
gdal.GDT_Int32, options=[
'TILED=YES', 'BIGTIFF=YES', 'COMPRESS=LZW',
'BLOCKXSIZE=16', 'BLOCKYSIZE=16'])
new_raster.SetProjection(projection_wkt)
new_raster.SetGeoTransform(geotransform)
new_band = new_raster.GetRasterBand(1)
new_band.SetNoDataValue(-1)
new_band.WriteArray(array)
new_raster.FlushCache()
new_band = None
new_raster = None
origin = (461261, 4923265)
pixel_size = (1, 1)
no_data = -1
pygeoprocessing.numpy_array_to_raster(
array, no_data, pixel_size, origin, projection_wkt,
base_raster_path)
class TestPreprocessor(unittest.TestCase):
"""Test Coastal Blue Carbon preprocessor functions."""
@ -1098,10 +1086,7 @@ class TestCBC2(unittest.TestCase):
"""Test `_calculate_npv`"""
from natcap.invest.coastal_blue_carbon import coastal_blue_carbon
def make_carbon_seq_raster(out_path):
"""make a carbon sequestration raster and save it to out_path"""
#make fake data
# make fake data
net_sequestration_rasters = {
2010: os.path.join(self.workspace_dir, "carbon_seq_2010.tif"),
2011: os.path.join(self.workspace_dir, "carbon_seq_2011.tif"),
@ -1140,10 +1125,9 @@ class TestCBC2(unittest.TestCase):
actual_2012 = band.ReadAsArray()
# compare actual rasters to expected (based on running `_calculate_npv`)
expected_2011 = numpy.array([[100550, 50300], [200950, 5000]],
dtype=int)
expected_2012 = numpy.array([[370268, 185195], [740045, 18409]],
dtype=int)
expected_2011 = numpy.array([[100525, 50262.5], [200950, 5000]])
expected_2012 = numpy.array([[370206.818182, 185103.409091],
[740045.454545, 18409.090909]])
numpy.testing.assert_allclose(actual_2011, expected_2011)
numpy.testing.assert_allclose(actual_2012, expected_2012)

12
tests/test_coastal_vulnerability.py
View File

@ -1560,7 +1560,7 @@ class CoastalVulnerabilityTests(unittest.TestCase):
from natcap.invest.coastal_vulnerability import \
assemble_results_and_calculate_exposure
def make_shore_points_vector(shore_points_path):
def _make_shore_points_vector(shore_points_path):
# create 4 points, each with a unique 'shore_id' in [0..3].
shore_geometries = [Point(0, 0), Point(1, 0), Point(2, 1), Point(3, 2)]
shore_fields = {'shore_id': ogr.OFTInteger}
@ -1577,7 +1577,7 @@ class CoastalVulnerabilityTests(unittest.TestCase):
ogr_geom_type=ogr.wkbPoint
)
def make_habitat_csv(habitat_csv_path):
def _make_habitat_csv(habitat_csv_path):
# Example: one habitat column named 'kelp', plus 'R_hab'
# We have 4 shore IDs, so we add 4 rows. Values are arbitrary.
habitat_df = pandas.DataFrame(
@ -1585,7 +1585,7 @@ class CoastalVulnerabilityTests(unittest.TestCase):
'seagrass': [4, 1, 2, 4], 'R_hab': [5, 2, 5, 3]})
habitat_df.to_csv(habitat_csv_path, index=False)
def make_risk_id_path_list():
def _make_risk_id_path_list():
# Create pickles for risk data
relief_pkl = os.path.join(self.workspace_dir, 'relief.pickle')
slr_pkl = os.path.join(self.workspace_dir, 'slr.pickle')
@ -1608,12 +1608,12 @@ class CoastalVulnerabilityTests(unittest.TestCase):
return risk_id_path_list
shore_points_path = os.path.join(self.workspace_dir, "shore_points.gpkg")
make_shore_points_vector(shore_points_path)
_make_shore_points_vector(shore_points_path)
habitat_csv_path = os.path.join(self.workspace_dir, 'habitat_protection.csv')
make_habitat_csv(habitat_csv_path)
_make_habitat_csv(habitat_csv_path)
risk_id_path_list = make_risk_id_path_list()
risk_id_path_list = _make_risk_id_path_list()
intermediate_vector_path = os.path.join(self.workspace_dir,
'intermediate_exposure.gpkg')

2
tests/test_crop_production.py
View File

@ -60,7 +60,7 @@ def make_simple_raster(base_raster_path, array):
Args:
base_raster_path (str): the raster path for making the new raster.
Returns:
Non
None.
"""
# UTM Zone 10N
srs = osr.SpatialReference()