diff --git a/.github/workflows/run-tests.yml b/.github/workflows/run-tests.yml index 4611307df..0181bbe1f 100644 --- a/.github/workflows/run-tests.yml +++ b/.github/workflows/run-tests.yml @@ -18,7 +18,7 @@ jobs: python -m pip install flake8==7.1.0 flake8-docstrings==1.7.0 flake8-annotations==3.1.1 - name: Run lint with flake8 run: | - flake8 src floodresilience + flake8 src floodresilience wrfhydro linting-pylint: name: Lint Python with pylint @@ -46,7 +46,7 @@ jobs: python -m pip install pylint==3.2.6 - name: Run lint with pylint run: | - pylint src floodresilience + pylint src floodresilience wrfhydro unit-tests: name: Run unit tests with pytest diff --git a/src/tasks.py b/src/tasks.py index 85c8718aa..ebe553c22 100644 --- a/src/tasks.py +++ b/src/tasks.py @@ -117,3 +117,4 @@ def wkt_to_gdf(wkt: str) -> gpd.GeoDataFrame: # These must be imported after app to remove a circular dependency import floodresilience.tasks # pylint: disable=wrong-import-position,unused-import # noqa: E402, F401 +import wrfhydro.tasks # pylint: disable=wrong-import-position,unused-import # noqa: E402, F401 diff --git a/wrfhydro/__init__.py b/wrfhydro/__init__.py new file mode 100644 index 000000000..7f03ad5e5 --- /dev/null +++ b/wrfhydro/__init__.py @@ -0,0 +1,17 @@ +# Copyright © 2021-2025 Geospatial Research Institute Toi Hangarau +# LICENSE: https://github.com/GeospatialResearch/Digital-Twins/blob/master/LICENSE +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Top-level package for the WRF-Hydro® extension for the Digital Twin.""" diff --git a/wrfhydro/blueprint.py b/wrfhydro/blueprint.py new file mode 100644 index 000000000..d1da1dfc7 --- /dev/null +++ b/wrfhydro/blueprint.py @@ -0,0 +1,30 @@ +"""Endpoints and flask configuration for running WRF-Hydro module within the digital twin""" + +from flask import Blueprint +from pywps import Service + +from wrfhydro.scenario.flood_scenario_process_service import FloodScenarioProcessService +from src.check_celery_alive import check_celery_alive + +wrf_hydro_blueprint = Blueprint('wrfhydro', __name__) +processes = [ + FloodScenarioProcessService() +] + +process_descriptor = {process.identifier: process.abstract for process in processes} + +service = Service(processes, ['src/pywps.cfg']) + + +@wrf_hydro_blueprint.route('/wps', methods=['GET', 'POST']) +@check_celery_alive +def wps() -> Service: + """ + End point for OGC WebProcessingService spec, allowing clients such as TerriaJS to request processing. + + Returns + ------- + Service + The PyWPS WebProcessing Service instance + """ + return service diff --git a/wrfhydro/forcing_data/__init__.py b/wrfhydro/forcing_data/__init__.py new file mode 100644 index 000000000..e9d292a89 --- /dev/null +++ b/wrfhydro/forcing_data/__init__.py @@ -0,0 +1,18 @@ +# -*- coding: utf-8 -*- +# Copyright © 2021-2025 Geospatial Research Institute Toi Hangarau +# LICENSE: https://github.com/GeospatialResearch/Digital-Twins/blob/master/LICENSE +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Package to handle data and operations specific to meteorological forcing data.""" diff --git a/wrfhydro/forcing_data/read_forcing_data.py b/wrfhydro/forcing_data/read_forcing_data.py new file mode 100644 index 000000000..824079044 --- /dev/null +++ b/wrfhydro/forcing_data/read_forcing_data.py @@ -0,0 +1,39 @@ +# # -*- coding: utf-8 -*- +# Copyright © 2021-2025 Geospatial Research Institute Toi Hangarau +# LICENSE: https://github.com/GeospatialResearch/Digital-Twins/blob/master/LICENSE +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Read meteorological forcing data into a form ready for use within WRF-Hydro modelling.""" + +import geopandas as gpd + +from src.digitaltwin.utils import LogLevel + + +def main( + selected_polygon_gdf: gpd.GeoDataFrame, + log_level: LogLevel = LogLevel.DEBUG +) -> None: + """ + Read meteorological forcing data into a form ready for use within WRF-Hydro modelling. + + Parameters + ---------- + selected_polygon_gdf : gpd.GeoDataFrame + A GeoDataFrame representing the selected polygon, i.e., the catchment area. + log_level : LogLevel = LogLevel.DEBUG + The log level to set for the root logger. Defaults to LogLevel.DEBUG. + """ + pass diff --git a/wrfhydro/land_surface_model/__init__.py b/wrfhydro/land_surface_model/__init__.py new file mode 100644 index 000000000..c4029ee4f --- /dev/null +++ b/wrfhydro/land_surface_model/__init__.py @@ -0,0 +1,18 @@ +# -*- coding: utf-8 -*- +# Copyright © 2021-2025 Geospatial Research Institute Toi Hangarau +# LICENSE: https://github.com/GeospatialResearch/Digital-Twins/blob/master/LICENSE +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Package to handle data and operations specific to forming land surface models.""" diff --git a/wrfhydro/land_surface_model/create_land_surface_model.py b/wrfhydro/land_surface_model/create_land_surface_model.py new file mode 100644 index 000000000..8eefa727e --- /dev/null +++ b/wrfhydro/land_surface_model/create_land_surface_model.py @@ -0,0 +1,39 @@ +# # -*- coding: utf-8 -*- +# Copyright © 2021-2025 Geospatial Research Institute Toi Hangarau +# LICENSE: https://github.com/GeospatialResearch/Digital-Twins/blob/master/LICENSE +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Create a land surface model for use within WRF-Hydro modelling.""" + +import geopandas as gpd + +from src.digitaltwin.utils import LogLevel + + +def main( + selected_polygon_gdf: gpd.GeoDataFrame, + log_level: LogLevel = LogLevel.DEBUG +) -> None: + """ + Create a land surface model for use within WRF-Hydro modelling. + + Parameters + ---------- + selected_polygon_gdf : gpd.GeoDataFrame + A GeoDataFrame representing the selected polygon, i.e., the catchment area. + log_level : LogLevel = LogLevel.DEBUG + The log level to set for the root logger. Defaults to LogLevel.DEBUG. + """ + pass diff --git a/wrfhydro/run_all.py b/wrfhydro/run_all.py new file mode 100644 index 000000000..3b56b0a1d --- /dev/null +++ b/wrfhydro/run_all.py @@ -0,0 +1,48 @@ +# -*- coding: utf-8 -*- +# Copyright © 2021-2025 Geospatial Research Institute Toi Hangarau +# LICENSE: https://github.com/GeospatialResearch/Digital-Twins/blob/master/LICENSE +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""This script runs each module in the Digital Twin using a Sample Polygon.""" +import pathlib + +from src.digitaltwin import retrieve_from_instructions +from src.digitaltwin.utils import LogLevel +from src.run_all import create_sample_polygon, main +from wrfhydro.forcing_data import read_forcing_data +from wrfhydro.land_surface_model import create_land_surface_model +from wrfhydro.scenario import run_wrf_hydro_scenario + +DEFAULT_MODULES_TO_PARAMETERS = { + retrieve_from_instructions: { + "log_level": LogLevel.INFO, + "instruction_json_path": pathlib.Path("wrfhydro/static_boundary_instructions.json").as_posix() + }, + create_land_surface_model: { + "log_level": LogLevel.INFO, + }, + read_forcing_data: { + "log_level": LogLevel.INFO, + }, + run_wrf_hydro_scenario: { + "log_level": LogLevel.INFO, + }, +} + +if __name__ == '__main__': + sample_polygon = create_sample_polygon() + + # Run all modules with sample polygon that intentionally contains slight rounding errors. + main(sample_polygon, DEFAULT_MODULES_TO_PARAMETERS) diff --git a/wrfhydro/scenario/__init__.py b/wrfhydro/scenario/__init__.py new file mode 100644 index 000000000..e9a6fb108 --- /dev/null +++ b/wrfhydro/scenario/__init__.py @@ -0,0 +1,18 @@ +# -*- coding: utf-8 -*- +# Copyright © 2021-2025 Geospatial Research Institute Toi Hangarau +# LICENSE: https://github.com/GeospatialResearch/Digital-Twins/blob/master/LICENSE +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Package to handle data and operations specific to running WRF-Hydro scenarios.""" diff --git a/wrfhydro/scenario/flood_scenario_process_service.py b/wrfhydro/scenario/flood_scenario_process_service.py new file mode 100644 index 000000000..34ecc061d --- /dev/null +++ b/wrfhydro/scenario/flood_scenario_process_service.py @@ -0,0 +1,153 @@ +# # -*- coding: utf-8 -*- +# Copyright © 2021-2025 Geospatial Research Institute Toi Hangarau +# LICENSE: https://github.com/GeospatialResearch/Digital-Twins/blob/master/LICENSE +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Defines PyWPS WebProcessingService process for creating a WRF-Hydro flooding scenario.""" + +import json + +from pywps import BoundingBoxInput, ComplexOutput, Format, Process, WPSRequest +from pywps.response.execute import ExecuteResponse +from shapely import box + +from wrfhydro import tasks +from src.config import EnvVariable as EnvVar + + +class FloodScenarioProcessService(Process): + """Class representing a WebProcessingService process for creating a flooding scenario""" + + # pylint: disable=too-few-public-methods + + def __init__(self) -> None: + """Define inputs and outputs of the WPS process, and assign process handler.""" + # Create bounding box WPS inputs + inputs = [ + BoundingBoxInput("bboxIn", "Area of Interest", crss=["epsg:4326"]), + ] + # Create flood inundation WPS outputs + outputs = [ + ComplexOutput("floodDepth", "Maximum Flood Depth", + supported_formats=[Format("application/vnd.terriajs.catalog-member+json")]), + ComplexOutput("floodedBuildings", "Flooded Buildings", + supported_formats=[Format("application/vnd.terriajs.catalog-member+json")]) + ] + + # Initialise the process + super().__init__( + self._handler, + identifier="wrfhydro", + title="Model a flood scenario using WRF-Hydro.", + inputs=inputs, + outputs=outputs, + ) + + @staticmethod + def _handler(request: WPSRequest, response: ExecuteResponse) -> None: + """ + Process handler for modelling a flood scenario + + Parameters + ---------- + request : WPSRequest + The WPS request, containing input parameters. + response : ExecuteResponse + The WPS response, containing output data. + """ + # Get coordinates from bounding box input + bounding_box_input = request.inputs['bboxIn'][0] + ymin, xmin = bounding_box_input.ll # lower left + ymax, xmax = bounding_box_input.ur # upper right + + # Form bounding box into standard shapely.box + bounding_box = box(xmin, ymin, xmax, ymax) + + modelling_task = tasks.create_scenario_for_area(bounding_box.wkt) + scenario_id = modelling_task.get() + + # Add Geoserver JSON Catalog entries to WPS response for use by Terria + response.outputs['floodDepth'].data = json.dumps(flood_depth_catalog(scenario_id)) + response.outputs['floodedBuildings'].data = json.dumps(building_flood_status_catalog(scenario_id)) + + +def building_flood_status_catalog(scenario_id: int) -> dict: + """ + Create a dictionary in the format of a terria js catalog json for the building flood status layer. + + Parameters + ---------- + scenario_id : int + The ID of the scenario to create the catalog item for. + + Returns + ---------- + dict + The TerriaJS catalog item JSON for the building flood status layer. + """ + dataset_name = "Building Flood Status" + gs_building_workspace = f"{EnvVar.POSTGRES_DB}-buildings" + gs_building_url = f"{EnvVar.GEOSERVER_HOST}:{EnvVar.GEOSERVER_PORT}/geoserver/{gs_building_workspace}/ows" + # Open and read HTML/mustache template file for infobox + return { + "type": "wfs", + "name": dataset_name, + "url": gs_building_url, + "typeNames": f"{gs_building_workspace}:building_flood_status", + "parameters": { + "viewparams": f"scenario:{scenario_id}", + }, + "maxFeatures": 300000, + "heightProperty": "extruded_height", + "legends": [{ + "title": "Building Flood Status", + "items": [ + { + "title": "Non-Flooded", + "color": "darkgreen" + }, + { + "title": "Flooded", + "color": "darkred" + } + ] + }] + } + + +def flood_depth_catalog(scenario_id: int) -> dict: + """ + Create a dictionary in the format of a terria js catalog json for the flood depth layer. + + Parameters + ---------- + scenario_id : int + The ID of the scenario to create the catalog item for. + + Returns + ---------- + dict + The TerriaJS catalog item JSON for the flood depth layer. + """ + gs_flood_model_workspace = f"{EnvVar.POSTGRES_DB}-wh-model-outputs" + gs_flood_url = f"{EnvVar.GEOSERVER_HOST}:{EnvVar.GEOSERVER_PORT}/geoserver/{gs_flood_model_workspace}/ows" + + return { + "type": "wms", + "name": "Flood Depth", + "url": gs_flood_url, + "layers": f"{gs_flood_model_workspace}:output_{scenario_id}", + "styles": "viridis_raster" + } diff --git a/wrfhydro/scenario/run_wrf_hydro_scenario.py b/wrfhydro/scenario/run_wrf_hydro_scenario.py new file mode 100644 index 000000000..47ca67213 --- /dev/null +++ b/wrfhydro/scenario/run_wrf_hydro_scenario.py @@ -0,0 +1,39 @@ +# # -*- coding: utf-8 -*- +# Copyright © 2021-2025 Geospatial Research Institute Toi Hangarau +# LICENSE: https://github.com/GeospatialResearch/Digital-Twins/blob/master/LICENSE +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Run a WRF-Hydro scenario for the selected polygon and serve outputs.""" + +import geopandas as gpd + +from src.digitaltwin.utils import LogLevel + + +def main( + selected_polygon_gdf: gpd.GeoDataFrame, + log_level: LogLevel = LogLevel.DEBUG +) -> None: + """ + Run a WRF-Hydro scenario for the selected polygon and serve outputs. + + Parameters + ---------- + selected_polygon_gdf : gpd.GeoDataFrame + A GeoDataFrame representing the selected polygon, i.e., the catchment area. + log_level : LogLevel = LogLevel.DEBUG + The log level to set for the root logger. Defaults to LogLevel.DEBUG. + """ + pass diff --git a/wrfhydro/static_boundary_instructions.json b/wrfhydro/static_boundary_instructions.json new file mode 100644 index 000000000..ccc8c6662 --- /dev/null +++ b/wrfhydro/static_boundary_instructions.json @@ -0,0 +1,16 @@ +{ + "instruction_coastlines": { + "data_provider": "LINZ", + "layer_id": 50258, + "table_name": "nz_coastlines", + "coverage_area": "New Zealand", + "url": "https://data.linz.govt.nz/layer/50258-nz-coastlines-topo-150k/" + }, + "instruction_buildings": { + "data_provider": "LINZ", + "layer_id": 101292, + "table_name": "nz_building_outlines", + "unique_column_name": "building_outline_id", + "url": "https://data.linz.govt.nz/layer/101292-nz-building-outlines-all-sources/" + } +} diff --git a/wrfhydro/tasks.py b/wrfhydro/tasks.py new file mode 100644 index 000000000..41b27d89f --- /dev/null +++ b/wrfhydro/tasks.py @@ -0,0 +1,106 @@ +""" +Runs backend tasks using Celery. Allowing for multiple long-running tasks to complete in the background. +Allows the frontend to send tasks and retrieve status later. +""" +import logging + +from celery import result, signals +from celery.worker.consumer import Consumer +import geopandas as gpd + +from src.digitaltwin import retrieve_from_instructions +from src.digitaltwin.utils import setup_logging +from src.tasks import add_base_data_to_db, app, OnFailureStateTask, wkt_to_gdf # pylint: disable=cyclic-import +from wrfhydro.forcing_data import read_forcing_data +from wrfhydro.land_surface_model import create_land_surface_model +from wrfhydro.scenario import run_wrf_hydro_scenario +from wrfhydro.run_all import DEFAULT_MODULES_TO_PARAMETERS + +setup_logging() +log = logging.getLogger(__name__) + + +@signals.worker_ready.connect +def on_startup(sender: Consumer, **_kwargs: None) -> None: # pylint: disable=missing-param-doc + """ + Initialise database, runs when Celery instance is ready. + + Parameters + ---------- + sender : Consumer + The Celery worker node instance + """ + with sender.app.connection() as conn: + # Gather area of interest from file. + aoi_wkt = gpd.read_file("selected_polygon.geojson").to_crs(4326).geometry[0].wkt + # Send a task to initialise this area of interest. + base_data_parameters = DEFAULT_MODULES_TO_PARAMETERS[retrieve_from_instructions] + sender.app.send_task("src.tasks.add_base_data_to_db", args=[aoi_wkt, base_data_parameters], connection=conn) + + +def create_scenario_for_area(selected_polygon_wkt: str) -> result.GroupResult: + """ + Create a model for the area using series of chained (sequential) sub-tasks. + + Parameters + ---------- + selected_polygon_wkt : str + The polygon defining the selected area to run the model for. Defined in WKT form. + + Returns + ------- + result.GroupResult + The task result for the long-running group of tasks. The task ID represents the final task in the group. + """ + base_data_parameters = DEFAULT_MODULES_TO_PARAMETERS[retrieve_from_instructions] + return ( + add_base_data_to_db.si(selected_polygon_wkt, base_data_parameters) | + create_land_surface_model_task.si(selected_polygon_wkt) | + read_forcing_data_task.si(selected_polygon_wkt) | + run_wrf_scenario_task.si(selected_polygon_wkt) + )() + + +@app.task(base=OnFailureStateTask) +def create_land_surface_model_task(selected_polygon_wkt: str) -> None: + """ + Task to ensure rainfall input data for the given area is added to the database and model input files are created. + + Parameters + ---------- + selected_polygon_wkt : str + The polygon defining the selected area to add rainfall data for. Defined in WKT form. + """ + parameters = DEFAULT_MODULES_TO_PARAMETERS[create_land_surface_model] + selected_polygon = wkt_to_gdf(selected_polygon_wkt) + create_land_surface_model.main(selected_polygon, **parameters) + + +@app.task(base=OnFailureStateTask) +def read_forcing_data_task(selected_polygon_wkt: str) -> None: + """ + Task to ensure meteorological forcing data is processed for the given area and added to the database. + + Parameters + ---------- + selected_polygon_wkt : str + The polygon defining the selected area to process the forcing data for. Defined in WKT form. + """ + parameters = DEFAULT_MODULES_TO_PARAMETERS[read_forcing_data] + selected_polygon = wkt_to_gdf(selected_polygon_wkt) + read_forcing_data.main(selected_polygon, **parameters) + + +@app.task(base=OnFailureStateTask) +def run_wrf_scenario_task(selected_polygon_wkt: str) -> None: + """ + Task to run a WRF Hydro scenario for a given area. + + Parameters + ---------- + selected_polygon_wkt : str + The polygon defining the selected area to run a wrf_hydro scenario for. Defined in WKT form. + """ + parameters = DEFAULT_MODULES_TO_PARAMETERS[run_wrf_hydro_scenario] + selected_polygon = wkt_to_gdf(selected_polygon_wkt) + run_wrf_hydro_scenario.main(selected_polygon, **parameters)