diff --git a/avaframe/in3Utils/spatialVoellmyInputs.py b/avaframe/in3Utils/spatialVoellmyInputs.py new file mode 100644 index 000000000..4ffad9347 --- /dev/null +++ b/avaframe/in3Utils/spatialVoellmyInputs.py @@ -0,0 +1,138 @@ +""" +Functions for generating spatial Voellmy friction raster inputs. +""" + +import pathlib +import logging +import numpy as np +import shapefile +from rasterio.features import rasterize +from shapely.geometry import shape, mapping + +from avaframe.in1Data.getInput import getAndCheckInputFiles, getDEMPath +from avaframe.in2Trans.rasterUtils import readRasterHeader, writeResultToRaster + +log = logging.getLogger(__name__) + + +def generateMuXsiRasters(avaDir, cfg): + """Generate mu and xi raster files from polygon shapefiles. + + Reads polygon shapefiles with "mu" and "xsi" attribute fields, + rasterizes the attribute values onto a grid matching the DEM extent + and resolution, and writes the rasters to Inputs/RASTERS/. + + Parameters + ---------- + avaDir : pathlib.Path + Path to avalanche directory containing Inputs/DEM and + Inputs/POLYGONS/ with *_mu.shp and *_xsi.shp shapefiles. + cfg : configparser.ConfigParser + Configuration with [DEFAULTS] section containing + default_mu and default_xsi values for uncovered areas. + """ + avaDir = pathlib.Path(avaDir) + inputDir = avaDir / "Inputs" + outDir = inputDir / "RASTERS" + outDir.mkdir(parents=True, exist_ok=True) + + # Find DEM + demPath = getDEMPath(avaDir) + demSuffix = demPath.suffix + + # Find shapefiles + muShpPath, muAvailable, _ = getAndCheckInputFiles( + inputDir, "POLYGONS", "mu shapefile", fileExt="shp", fileSuffix="_mu" + ) + if muAvailable == "No": + raise FileNotFoundError("No *_mu.shp found in %s/POLYGONS/" % inputDir) + xsiShpPath, xsiAvailable, _ = getAndCheckInputFiles( + inputDir, "POLYGONS", "xsi shapefile", fileExt="shp", fileSuffix="_xsi" + ) + if xsiAvailable == "No": + raise FileNotFoundError("No *_xsi.shp found in %s/POLYGONS/" % inputDir) + + # Read DEM header + demHeader = readRasterHeader(demPath) + demTransform = demHeader["transform"] + demCrs = demHeader["crs"] + demShape = (demHeader["nrows"], demHeader["ncols"]) + + defaultMu = cfg["DEFAULTS"].getfloat("default_mu") + defaultXsi = cfg["DEFAULTS"].getfloat("default_xsi") + + # Rasterize mu + log.info("Rasterizing mu shapefile: %s", muShpPath) + muRaster = _rasterizeShapefile(muShpPath, defaultMu, "mu", demShape, demTransform) + + # Rasterize xsi + log.info("Rasterizing xsi shapefile: %s", xsiShpPath) + xsiRaster = _rasterizeShapefile(xsiShpPath, defaultXsi, "xsi", demShape, demTransform) + + # Determine output driver + if demSuffix == ".asc": + driver = "AAIGrid" + else: + driver = "GTiff" + + # Write output + outHeader = { + "driver": driver, + "crs": demCrs, + "transform": demTransform, + "nodata_value": None, + } + log.info("Writing mu raster") + writeResultToRaster(outHeader, muRaster, outDir / "raster_mu") + log.info("Writing xsi raster") + writeResultToRaster(outHeader, xsiRaster, outDir / "raster_xi") + log.info("Raster generation completed.") + + +def _rasterizeShapefile(shpPath, defaultValue, fieldName, demShape, demTransform): + """Rasterize a polygon shapefile attribute field onto a DEM-matching grid. + + All cells not covered by any polygon are filled with defaultValue. + + Parameters + ---------- + shpPath : pathlib.Path + Path to shapefile. + defaultValue : float + Fill value for cells not covered by any polygon. + fieldName : str + Attribute field name to extract from each feature. + demShape : tuple + (height, width) of the output raster. + demTransform : affine.Affine + Geotransform of the DEM. + + Returns + ------- + raster : numpy.ndarray + Rasterized array with shape demShape. + """ + with shapefile.Reader(str(shpPath)) as sf: + fieldNames = [f[0].lower() for f in sf.fields[1:]] + if fieldName not in fieldNames: + raise KeyError( + "Field '%s' not found in %s. Available fields: %s" % (fieldName, shpPath.name, fieldNames) + ) + fieldIdx = fieldNames.index(fieldName) + + shapes = [] + for rec in sf.shapeRecords(): + geom = rec.shape.__geo_interface__ + poly = shape(geom) + value = rec.record[fieldIdx] + shapes.append((mapping(poly), value)) + + raster = rasterize( + shapes, + out_shape=demShape, + transform=demTransform, + fill=defaultValue, + all_touched=True, + dtype=np.float32, + ) + return raster diff --git a/avaframe/in3Utils/spatialVoellmyInputsCfg.ini b/avaframe/in3Utils/spatialVoellmyInputsCfg.ini new file mode 100644 index 000000000..1fb927e33 --- /dev/null +++ b/avaframe/in3Utils/spatialVoellmyInputsCfg.ini @@ -0,0 +1,6 @@ +[DEFAULTS] +# Default mu value for areas not covered by shapefiles +default_mu = 0.155 + +# Default xsi value for areas not covered by shapefiles +default_xsi = 4000. diff --git a/avaframe/runScripts/runSpatialVoellmyInputs.py b/avaframe/runScripts/runSpatialVoellmyInputs.py new file mode 100644 index 000000000..f02c89ad8 --- /dev/null +++ b/avaframe/runScripts/runSpatialVoellmyInputs.py @@ -0,0 +1,55 @@ +""" +Run script for generating spatial Voellmy friction raster inputs. +""" + +import argparse +import pathlib +import time + +from avaframe.in3Utils import cfgUtils +from avaframe.in3Utils import logUtils +from avaframe.in3Utils import initializeProject as initProj +from avaframe.in3Utils import spatialVoellmyInputs + + +def runSpatialVoellmyInputs(avaDir=""): + """Run generation of mu and xi rasters from shapefiles. + + Parameters + ---------- + avaDir : str + Path to the avalanche directory. If empty, read from + avaframeCfg.ini. + """ + startTime = time.time() + cfgMain = cfgUtils.getGeneralConfig() + if avaDir: + avaDir = pathlib.Path(avaDir) + else: + avaDir = pathlib.Path(cfgMain["MAIN"]["avalancheDir"]) + + logName = "runSpatialVoellmyInputs" + log = logUtils.initiateLogger(avaDir, logName) + log.info("MAIN SCRIPT") + log.info("Avalanche directory: %s", avaDir) + + initProj.cleanSingleAvaDir(avaDir, deleteOutput=False) + + cfg = cfgUtils.getModuleConfig(spatialVoellmyInputs) + spatialVoellmyInputs.generateMuXsiRasters(avaDir, cfg) + + endTime = time.time() + log.info("Took %6.1f seconds to calculate.", endTime - startTime) + log.info("Workflow completed successfully.") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Generate spatial Voellmy mu/xi rasters from shapefiles") + parser.add_argument( + "avaDir", + nargs="?", + default="", + help="Path to avalanche directory", + ) + args = parser.parse_args() + runSpatialVoellmyInputs(str(args.avaDir)) diff --git a/avaframe/tests/test_spatialVoellmyInputs.py b/avaframe/tests/test_spatialVoellmyInputs.py new file mode 100644 index 000000000..6b6f0b9ae --- /dev/null +++ b/avaframe/tests/test_spatialVoellmyInputs.py @@ -0,0 +1,204 @@ +"""Tests for module spatialVoellmyInputs""" + +import pathlib +import tempfile +import shutil +import numpy as np +import rasterio +import shapefile +import pytest +from avaframe.in3Utils import spatialVoellmyInputs + + +def _makeSyntheticDEM(tmpDir, suffix=".asc"): + """Create a small DEM raster for testing. + + Returns path to DEM, its transform, crs, and shape. + """ + demPath = tmpDir / f"DEM{suffix}" + data = np.arange(100, dtype=np.float32).reshape(10, 10) + # North-up: 10x10 grid from (0,0) to (10,10) + transform = rasterio.transform.from_bounds(0, 0, 10, 10, 10, 10) + crs = "EPSG:32633" + driver = "AAIGrid" if suffix == ".asc" else "GTiff" + with rasterio.open( + demPath, + "w", + driver=driver, + height=10, + width=10, + count=1, + dtype=data.dtype, + crs=crs, + transform=transform, + ) as dst: + dst.write(data, 1) + return demPath, transform, crs, data.shape + + +def _makeSyntheticShapefile(shpPath, fieldName, featureCoordsValues): + """Create a shapefile with a single field and polygon features. + + featureCoordsValues: list of (coords_list, field_value) tuples. + coords_list is list of (x, y) tuples forming a clockwise ring. + """ + with shapefile.Writer(shpPath, shapeType=shapefile.POLYGON) as w: + w.field(fieldName, "F", decimal=6) + for coords, value in featureCoordsValues: + w.poly([coords]) + w.record(value) + + +def test_generateMuXsiRasters_asc(): + """Test raster generation with .asc DEM and shapefiles.""" + tmpDir = pathlib.Path(tempfile.mkdtemp()) + try: + # Setup: DEM + demPath, transform, crs, demShape = _makeSyntheticDEM(tmpDir, ".asc") + inputsDir = tmpDir / "Inputs" + inputsDir.mkdir() + shutil.move(str(demPath), str(inputsDir / "DEM.asc")) + demPath = inputsDir / "DEM.asc" + + # Setup: mu shapefile with two polygons + # Polygon 1: geographic (2..5, 7..9) -> rows 1-3, cols 2-5 + # Polygon 2: geographic (6..9, 2..5) -> rows 5-8, cols 6-9 + polyDir = inputsDir / "POLYGONS" + polyDir.mkdir() + muShp = polyDir / "zones_mu.shp" + _makeSyntheticShapefile( + muShp, + "mu", + [ + ([(2, 7), (5, 7), (5, 9), (2, 9), (2, 7)], 0.300), + ([(6, 2), (9, 2), (9, 5), (6, 5), (6, 2)], 0.500), + ], + ) + + # Setup: xsi shapefile (same geometry, different values) + xsiShp = polyDir / "zones_xsi.shp" + _makeSyntheticShapefile( + xsiShp, + "xsi", + [ + ([(2, 7), (5, 7), (5, 9), (2, 9), (2, 7)], 3000.0), + ([(6, 2), (9, 2), (9, 5), (6, 5), (6, 2)], 5000.0), + ], + ) + + # Setup: config + import configparser + + cfg = configparser.ConfigParser() + cfg["DEFAULTS"] = {"default_mu": "0.155", "default_xsi": "4000."} + + # Run + spatialVoellmyInputs.generateMuXsiRasters(tmpDir, cfg) + + # Assert output files exist + rastersDir = inputsDir / "RASTERS" + muRasterPath = rastersDir / "raster_mu.asc" + xsiRasterPath = rastersDir / "raster_xi.asc" + assert muRasterPath.exists() + assert xsiRasterPath.exists() + + # Assert mu raster values + with rasterio.open(muRasterPath) as src: + muData = src.read(1) + assert src.transform == transform + assert src.shape == demShape + # Covered cells (interior of polygons) + assert muData[2, 3] == pytest.approx(0.300) # poly 1: row 2, col 3 + assert muData[6, 7] == pytest.approx(0.500) # poly 2: row 6, col 7 + # Uncovered cell should have default + assert muData[9, 0] == pytest.approx(0.155) # row 9, col 0 outside + + # Assert xsi raster values + with rasterio.open(xsiRasterPath) as src: + xsiData = src.read(1) + assert xsiData[2, 3] == pytest.approx(3000.0) + assert xsiData[6, 7] == pytest.approx(5000.0) + assert xsiData[9, 0] == pytest.approx(4000.0) + + finally: + shutil.rmtree(tmpDir) + + +def test_generateMuXsiRasters_tif(): + """Test raster generation with .tif DEM -- output should be .tif.""" + tmpDir = pathlib.Path(tempfile.mkdtemp()) + try: + demPath, transform, crs, demShape = _makeSyntheticDEM(tmpDir, ".tif") + inputsDir = tmpDir / "Inputs" + inputsDir.mkdir() + shutil.move(str(demPath), str(inputsDir / "DEM.tif")) + + polyDir = inputsDir / "POLYGONS" + polyDir.mkdir() + muShp = polyDir / "zones_mu.shp" + _makeSyntheticShapefile(muShp, "mu", [([(2, 7), (5, 7), (5, 9), (2, 9), (2, 7)], 0.300)]) + xsiShp = polyDir / "zones_xsi.shp" + _makeSyntheticShapefile(xsiShp, "xsi", [([(2, 7), (5, 7), (5, 9), (2, 9), (2, 7)], 3000.0)]) + + import configparser + + cfg = configparser.ConfigParser() + cfg["DEFAULTS"] = {"default_mu": "0.155", "default_xsi": "4000."} + + spatialVoellmyInputs.generateMuXsiRasters(tmpDir, cfg) + + rastersDir = inputsDir / "RASTERS" + muPath = rastersDir / "raster_mu.tif" + xsiPath = rastersDir / "raster_xi.tif" + assert muPath.exists() + assert xsiPath.exists() + assert muPath.suffix == ".tif" + assert xsiPath.suffix == ".tif" + + finally: + shutil.rmtree(tmpDir) + + +def test_missingMuFieldRaises(): + """Test that missing 'mu' field in shapefile raises clear error.""" + tmpDir = pathlib.Path(tempfile.mkdtemp()) + try: + demPath, _, _, _ = _makeSyntheticDEM(tmpDir, ".asc") + inputsDir = tmpDir / "Inputs" + inputsDir.mkdir() + shutil.move(str(demPath), str(inputsDir / "DEM.asc")) + polyDir = inputsDir / "POLYGONS" + polyDir.mkdir() + # Shapefile with wrong field name + _makeSyntheticShapefile( + polyDir / "zones_mu.shp", "friction_mu", [([(2, 7), (5, 7), (5, 9), (2, 9), (2, 7)], 0.3)] + ) + _makeSyntheticShapefile( + polyDir / "zones_xsi.shp", "xsi", [([(2, 7), (5, 7), (5, 9), (2, 9), (2, 7)], 3000.0)] + ) + + import configparser + + cfg = configparser.ConfigParser() + cfg["DEFAULTS"] = {"default_mu": "0.1", "default_xsi": "300."} + + with pytest.raises(KeyError, match="mu"): + spatialVoellmyInputs.generateMuXsiRasters(tmpDir, cfg) + finally: + shutil.rmtree(tmpDir) + + +def test_missingDEMRaises(): + """Test that missing DEM raises an error.""" + tmpDir = pathlib.Path(tempfile.mkdtemp()) + try: + inputsDir = tmpDir / "Inputs" + inputsDir.mkdir() + import configparser + + cfg = configparser.ConfigParser() + cfg["DEFAULTS"] = {"default_mu": "0.1", "default_xsi": "300."} + with pytest.raises(FileNotFoundError): + spatialVoellmyInputs.generateMuXsiRasters(tmpDir, cfg) + finally: + shutil.rmtree(tmpDir) diff --git a/docs/moduleCom1DFA.rst b/docs/moduleCom1DFA.rst index 31ad44898..b38798909 100644 --- a/docs/moduleCom1DFA.rst +++ b/docs/moduleCom1DFA.rst @@ -107,6 +107,7 @@ at least two results are generated: the *null* variant and the variant with entr - only one file per parameter allowed - if ``meshCellSize`` is different from simulation ``meshCellSize`` fields will be remeshed - only used if ``frictionModel`` is set to ``spatialVoellmy`` + - to generate these rasters from polygon shapefiles, see :ref:`moduleIn3Utils:Spatial Voellmy inputs` * **one ``_cropshape.shp`` shape file (in Inputs/POLYGONS)** diff --git a/docs/moduleIn3Utils.rst b/docs/moduleIn3Utils.rst index e16e51620..4c5d38286 100644 --- a/docs/moduleIn3Utils.rst +++ b/docs/moduleIn3Utils.rst @@ -10,6 +10,37 @@ comparison or interpolation on rasters, lines, points... Further information about the available functions can be found in :py:mod:`in3Utils.geoTrans` +Spatial Voellmy inputs +====================== + +The :py:mod:`in3Utils.spatialVoellmyInputs` module generates raster files +for the Voellmy friction parameters :math:`\mu` and :math:`\xi` from polygon +shapefiles. This is required when using the ``spatialVoellmy`` friction model +in com1DFA (see :ref:`moduleCom1DFA:Input`). + +Provide polygon shapefiles with ``mu`` and ``xsi`` attribute fields in +``Inputs/POLYGONS/``, with file names ending in ``_mu.shp`` and ``_xsi.shp``. +The DEM must be placed in ``Inputs/``. The generated rasters are written to +``Inputs/RASTERS/`` with the same file format as the DEM. + +Default values for areas not covered by the polygon shapefiles are set in +``avaframe/in3Utils/spatialVoellmyInputsCfg.ini``. + + +To run +------ + +* first go to ``AvaFrame/avaframe`` +* copy ``in3Utils/spatialVoellmyInputsCfg.ini`` to + ``in3Utils/local_spatialVoellmyInputsCfg.ini`` and set desired + ``default_mu`` and ``default_xsi`` values (if not, the default values + are used) +* ensure the DEM and shapefiles are in the avalanche directory as + described above +* run:: + + python3 runScripts/runSpatialVoellmyInputs.py + Generate Topography ===================