# -*- coding: utf-8 -*-
"""
* Copyright (C) 2023-2025 Alexandre Gauvain, Ronan Abhervé, Jean-Raynald de Dreuzy
*
* This program and the accompanying materials are made available under the
* terms of the Eclipse Public License 2.0 which is available at
* http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
* which is available at https://www.apache.org/licenses/LICENSE-2.0.
*
* SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
"""
#%% LIBRAIRIES
# Python
import os
import pandas as pd
import geopandas as gpd
import numpy as np
import rasterio
import whitebox
from hydromodpy.tools import get_logger
wbt = whitebox.WhiteboxTools()
wbt.verbose = False
logger = get_logger(__name__)
#%% CLASS
[docs]
class Hydrography:
"""
Add hydrography data in the watershed object.
"""
def __init__(self, out_path: str,
types_obs: list, fields_obs: list,
geographic: object, hydro_path: str,
streams_file=None):
"""
Parameters
----------
out_path : str
Path of the HydroModPy outputs.
types_obs : str
Label of the shpaefile .shp file.
fields_obs : str
Column field label rasterized from the shapefile.
geographic : object
Variable object of the model domain (watershed).
hydro_path : str
Path of the folder with the hydrography data.
"""
logger.info("Extracting hydrography data from %s", hydro_path)
data_folder = out_path + '/results_stable/hydrography/'
if not os.path.exists(data_folder):
os.makedirs(data_folder)
self.hydro_path = hydro_path
watershed_shp = geographic.watershed_shp # watershed_shp = geographic.watershed_box_shp
watershed_dem = geographic.watershed_dem # watershed_dem = geographic.watershed_box_buff_dem
for type_obs, field_obs in zip(types_obs, fields_obs):
try:
self.clip_observed(type_obs, field_obs, hydro_path, data_folder, watershed_shp, watershed_dem, streams_file)
except ValueError as e:
logger.error("Hydrography extraction failed for %s: %s", type_obs, e)
pass
#%% FUNCTIONS
[docs]
def clip_observed(self, type_obs, field_obs, hydro_path, data_folder, watershed_shp, watershed_dem, streams_file):
"""
Function to clip hydrogrpahic data at the watershed scale (or model domain).
Parameters
----------
watershed_shp : str
Path of the watershed shapefile (model domain) generated by geographic.
watershed_dem : str
Path of the watershed raster (model domain) generated by geographic.
"""
streams = hydro_path + '/' + type_obs +'.shp'
self.streams = data_folder + type_obs +'.shp'
# First clip of the shape file at the watershed scale (classical GIS function performed here in geopandas)
if isinstance(streams_file, gpd.GeoDataFrame):
streams_file = streams_file
else:
streams_file = gpd.read_file(streams)
watshd_file = gpd.read_file(watershed_shp)
file_clipped = gpd.clip(streams_file, watshd_file) # wbt.clip(streams, watershed_shp, self.streams)
# Saves clipped file to the reuslts file structure
file_clipped.to_file(self.streams)
# Transforms shapefile to raster file (.tif format)
shp_base = gpd.read_file(self.streams)
shp_type = shp_base.geometry.type[0] # forma = forma.geom_type[0]
self.tif_streams = data_folder + type_obs + '.tif'
try:
shp_base[field_obs] = pd.to_numeric(shp_base[field_obs])
except:
pass
shp_base.to_file(self.streams)
if (shp_type == 'MultiPolygon') | (shp_type == 'Polygon'): # if shp_type == 'LineString':
logger.debug('Processing polygon geometry type: %s', shp_type)
# e.g. wetlands and ponds
# wbt.dissolve(self.streams, self.streams)
wbt.vector_polygons_to_raster(self.streams, self.tif_streams, field=field_obs, base=watershed_dem)
if (shp_type == 'MultiLineString') | (shp_type == 'LineString') | (shp_type == 'Line'):
logger.debug('Processing line geometry type: %s', shp_type)
# e.g. streams
wbt.vector_lines_to_raster(self.streams, self.tif_streams, field=field_obs,base=watershed_dem)
if (shp_type == 'Point') | (shp_type == 'MultiPoint') :
logger.debug('Processing point geometry type: %s', shp_type)
# e.g. landslides, sources, wells
wbt.vector_points_to_raster(self.streams, self.tif_streams, field=field_obs, base=watershed_dem)
wbt.set_nodata_value(
self.tif_streams,
self.tif_streams,
back_value=-32768)
with rasterio.open(self.tif_streams) as dem_streams:
self.streams_array = dem_streams.read(1).astype(float)
self.streams_array[self.streams_array<0] = np.nan
pt_streams = data_folder + type_obs + '_pt.shp'
wbt.raster_to_vector_points(self.tif_streams, pt_streams)
#%% NOTES
