#2600

Committed 01 Oct 2025 08:42AM UTC coverage: 34.793% (-0.3%) from 35.131%

Build # #2600

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Fat improvements (#1143)

* qt6 stub

* qt6

* qt5 -> qt6

* convert

* fix

* FAT: Toggle items on/off with space bar

* basic color changing, still need to save in project

* lint

* Switching stacked or volume mode takes selection into account

* Fraction Analysis Tool: substances are listed alphabetically (#1133)

* CHANGES

* Qt6 fixes

* Plot highlighting (#1132)

* Stub plot highlighting

* stub plot highlighting

* Optimizations

* stub value marker

* Closest curve is now pixel-based

* Make coverals non-blocking for PRs GA

* optimiziation

* Proper highlighting of fills

* proper highlighting of bottom fill (uses fillLevel)

* lint

* refactor

* FAT: curve marker has same color as closest line

Run Details

82 of 406 new or added lines in 42 files covered. (20.2%)

9 existing lines in 6 files now uncovered.

4878 of 14020 relevant lines covered (34.79%)

0.35 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

32.47

/processing/leak_detector_algorithms.py

# -*- coding: utf-8 -*-

"""
***************************************************************************
*                                                                         *
*   This program is free software; you can redistribute it and/or modify  *
*   it under the terms of the GNU General Public License as published by  *
*   the Free Software Foundation; either version 2 of the License, or     *
*   (at your option) any later version.                                   *
*                                                                         *
***************************************************************************
"""
from pathlib import Path
from osgeo import gdal
from typing import Any, Dict, Tuple, Iterator, List
from shapely import wkt
from threedigrid.admin.gridadmin import GridH5Admin
from threedigrid.admin.gridresultadmin import GridH5ResultAdmin
from qgis.PyQt.QtCore import QVariant
from qgis.core import (
    QgsCoordinateReferenceSystem,
    QgsGeometry,
    QgsFeature,
    QgsFeatureSink,
    QgsField,
    QgsFields,
    QgsProcessing,
    QgsProcessingAlgorithm,
    QgsProcessingContext,
    QgsProcessingException,
    QgsProcessingParameterFeatureSink,
    QgsProcessingParameterFeatureSource,
    QgsProcessingParameterFile,
    QgsProcessingParameterNumber,
    QgsProcessingParameterRasterLayer,
    QgsWkbTypes,
)

from threedi_results_analysis.processing.deps.discharge.leak_detector import LeakDetector
from threedi_results_analysis.processing.deps.discharge.discharge_reduction import LeakDetectorWithDischargeThreshold
from threedi_results_analysis.utils.threedi_result_aggregation.aggregation_classes import Aggregation, AggregationSign
from threedi_results_analysis.utils.threedi_result_aggregation.constants import AGGREGATION_VARIABLES, AGGREGATION_METHODS

Q_NET_SUM = Aggregation(
    variable=AGGREGATION_VARIABLES.get_by_short_name("q"),
    method=AGGREGATION_METHODS.get_by_short_name("sum"),
    sign=AggregationSign("net", "Net"),
)

STYLE_DIR = Path(__file__).parent / "styles"

QVARIANT_PYTHON_TYPES = {
    QVariant.Int: int,
    QVariant.Double: float
}


class DetectLeakingObstaclesBase(QgsProcessingAlgorithm):
    """
    Base algorithm for 'Detect leaking obstacles' algorithms, not to be exposed to users
    """

    INPUT_GRIDADMIN = "INPUT_GRIDADMIN"
    INPUT_DEM = "INPUT_DEM"
    INPUT_FLOWLINES = "INPUT_FLOWLINES"
    INPUT_OBSTACLES = "INPUT_OBSTACLES"
    INPUT_MIN_OBSTACLE_HEIGHT = "INPUT_MIN_OBSTACLE_HEIGHT"

    OUTPUT_EDGES = "OUTPUT_EDGES"
    OUTPUT_OBSTACLES = "OUTPUT_OBSTACLES"

    def initAlgorithm(self, config):

        self.addParameter(
            QgsProcessingParameterFile(
                self.INPUT_GRIDADMIN, "Gridadmin file", extension="h5"
            )
        )

        self.addParameter(
            QgsProcessingParameterRasterLayer(
                self.INPUT_DEM, "Digital Elevation Model"
            )
        )

        self.addParameter(
            QgsProcessingParameterFeatureSource(
                self.INPUT_OBSTACLES,
                "Linear obstacles",
                [QgsProcessing.SourceType.TypeVectorLine],
                optional=True
            )
        )

        self.addParameter(
            QgsProcessingParameterFeatureSource(
                self.INPUT_FLOWLINES,
                "Flowlines",
                [QgsProcessing.SourceType.TypeVectorLine],
                optional=True
            )
        )

        min_obstacle_height_param = QgsProcessingParameterNumber(
            self.INPUT_MIN_OBSTACLE_HEIGHT,
            "Minimum obstacle height (m)",
            type=QgsProcessingParameterNumber.Type.Double
        )
        min_obstacle_height_param.setMetadata({"widget_wrapper": {"decimals": 3}})
        self.addParameter(min_obstacle_height_param)

        self.addParameter(
            QgsProcessingParameterFeatureSink(
                self.OUTPUT_EDGES,
                "Output: Obstacle on cell edge",
                type=QgsProcessing.SourceType.TypeVectorLine
            )
        )

        self.addParameter(
            QgsProcessingParameterFeatureSink(
                self.OUTPUT_OBSTACLES,
                "Output: Obstacle in DEM",
                type=QgsProcessing.SourceType.TypeVectorLine
            )
        )

    def checkParameterValues(self, parameters: Dict[str, Any], context: QgsProcessingContext) -> Tuple[bool, str]:
        success, msg = super().checkParameterValues(parameters, context)
        if success:
            msg_list = list()

            # check if min_obstacle_height > 0
            min_obstacle_height = self.parameterAsDouble(parameters, self.INPUT_MIN_OBSTACLE_HEIGHT, context)
            if min_obstacle_height <= 0:
                msg_list.append('Minimum obstacle height must be greater than 0')

            # check input obstacles has a field called crest_level
            input_obstacles_source = self.parameterAsSource(parameters, self.INPUT_OBSTACLES, context)
            if input_obstacles_source:
                if 'crest_level' not in input_obstacles_source.fields().names():
                    msg_list.append('Obstacle lines layer does not contain crest_level field')
            success = len(msg_list) == 0
            msg = '; '.join(msg_list)
        return success, msg

    @staticmethod
    def sink_field_data() -> List[Dict]:
        """
        Return a list of dicts that contain the data needed to defined fields in the feature sinks (output layers).
        Each dict contains the name (str) and the type (QVariant)
        """
        return [
            {"name": "flowline_id", "type": QVariant.Int},
            {"name": "exchange_level", "type": QVariant.Double},
            {"name": "crest_level", "type": QVariant.Double}
        ]

    def add_features_to_sink(self, feedback, sink: QgsFeatureSink, features_data: Iterator):
        for i, feature_data in enumerate(features_data):
            if feedback.isCanceled():
                return {}
            feature = QgsFeature()
            feature.setFields(self.sink_fields)
            for i, field in enumerate(self.sink_field_data()):
                convert = QVARIANT_PYTHON_TYPES[field["type"]]
                feature.setAttribute(i, convert(feature_data[field["name"]]))
            geometry = QgsGeometry()
            geometry.fromWkb(feature_data["geometry"].wkb)
            feature.setGeometry(geometry)
            sink.addFeature(feature, QgsFeatureSink.Flag.FastInsert)

    def read_parameters(self, parameters, context, feedback):
        self.gridadmin_fn = self.parameterAsFile(parameters, self.INPUT_GRIDADMIN, context)
        self.gridadmin = GridH5Admin(self.gridadmin_fn)
        dem = self.parameterAsRasterLayer(parameters, self.INPUT_DEM, context)
        dem_fn = dem.dataProvider().dataSourceUri()
        self.dem_ds = gdal.Open(dem_fn)
        flowlines_source = self.parameterAsSource(parameters, self.INPUT_FLOWLINES, context)
        obstacles_source = self.parameterAsSource(parameters, self.INPUT_OBSTACLES, context)
        self.min_obstacle_height = self.parameterAsDouble(parameters, self.INPUT_MIN_OBSTACLE_HEIGHT, context)

        crs = QgsCoordinateReferenceSystem(f"EPSG:{self.gridadmin.epsg_code}")

        self.sink_fields = QgsFields()
        for field_data in self.sink_field_data():
            self.sink_fields.append(QgsField(**field_data))

        self.edges_sink, self.edges_sink_dest_id = self.parameterAsSink(
            parameters,
            self.OUTPUT_EDGES,
            context,
            fields=self.sink_fields,
            geometryType=QgsWkbTypes.Type.LineString,
            crs=crs
        )
        self.obstacles_sink, self.obstacles_sink_dest_id = self.parameterAsSink(
            parameters,
            self.OUTPUT_OBSTACLES,
            context,
            fields=self.sink_fields,
            geometryType=QgsWkbTypes.Type.LineString,
            crs=crs
        )

        # get list of flowline ids
        if flowlines_source:
            field_index = flowlines_source.fields().indexFromName('id')
            self.flowline_ids = [feature.attributes()[field_index] for feature in flowlines_source.getFeatures()]
        else:
            self.flowline_ids = list(self.gridadmin.lines.id)

        if obstacles_source:
            feedback.setProgressText("Read linear obstacles input...")
            if obstacles_source.sourceCrs() != crs:
                raise QgsProcessingException(
                    "Obstacles input has different Coordinate Reference System than the gridadmin file"
                )
            crest_level_field_idx = obstacles_source.fields().indexFromName("crest_level")

            self.input_obstacles = list()
            for input_obstacle in obstacles_source.getFeatures():
                geom = wkt.loads(input_obstacle.geometry().asWkt())
                crest_level = float(input_obstacle[crest_level_field_idx])
                self.input_obstacles.append((geom, crest_level))
        else:
            self.input_obstacles = None

    def get_leak_detector(self, feedback):
        leak_detector = LeakDetector(
            gridadmin=self.gridadmin,
            dem=self.dem_ds,
            flowline_ids=self.flowline_ids,
            min_obstacle_height=self.min_obstacle_height,
            obstacles=self.input_obstacles,
            feedback=feedback
        )
        return leak_detector

    def processAlgorithm(self, parameters, context, feedback):
        self.read_parameters(parameters, context, feedback)
        feedback.setProgressText("Read computational grid...")
        leak_detector = self.get_leak_detector(feedback)
        if feedback.isCanceled():
            return {}
        feedback.setProgressText("Find obstacles...")
        leak_detector.run(feedback=feedback)
        feedback.setProgressText("Create 'Obstacle on cell edge' features...")
        self.add_features_to_sink(
            feedback=feedback,
            sink=self.edges_sink,
            features_data=leak_detector.results(geometry='EDGE')
        )
        feedback.setProgressText("Create 'Obstacle in DEM' features...")
        self.add_features_to_sink(
            feedback=feedback,
            sink=self.obstacles_sink,
            features_data=leak_detector.results(geometry='OBSTACLE')
        )

        return {
            self.OUTPUT_EDGES: self.edges_sink_dest_id,
            self.OUTPUT_OBSTACLES: self.obstacles_sink_dest_id
        }

    def group(self):
        return "Computational Grid"

    def groupId(self):
        return "computational_grid"


class DetectLeakingObstaclesAlgorithm(DetectLeakingObstaclesBase):
    """
    Detect obstacle lines in the DEM that are ignored by 3Di due to its location relative to cell edges
    """
    def postProcessAlgorithm(self, context, feedback):
        """Set styling of output vector layers"""
        edges_layer = context.getMapLayer(self.edges_sink_dest_id)
        edges_layer.loadNamedStyle(str(STYLE_DIR / "obstacle_on_cell_edge.qml"))

        obstacles_layer = context.getMapLayer(self.obstacles_sink_dest_id)
        obstacles_layer.loadNamedStyle(str(STYLE_DIR / "obstacle_in_dem.qml"))

        return {
            self.OUTPUT_EDGES: self.edges_sink_dest_id,
            self.OUTPUT_OBSTACLES: self.obstacles_sink_dest_id
        }

    def name(self):
        """
        Returns the algorithm name, used for identifying the algorithm
        """
        return "detect_leaking_obstacles"

    def displayName(self):
        """
        Returns the algorithm name, which should be used for any
        user-visible display of the algorithm name.
        """
        return "Detect leaking obstacles in DEM"

    def shortHelpString(self):
        return """
                <h3>Introduction</h3>
                <p>The elevation at which flow between 2D cells is possible (the 'exchange level'), depends on the elevation of the pixels directly adjacent to the cell edge. Obstacles in the DEM that do not cover the entire edge will therefore not stop the flow, i.e. water 'leaks' through the obstacle. This is more likely to occur if obstacles are diagonal and/or narrow compared to the computational grid size.</p>
                <p>This processing algorithm detects such cases. Please inspect the locations where the algorithm identifies leaking obstacles and add grid refinement and/or obstacles to the schematisation to solve the issue if needed.</p>
                <h3>Parameters</h3>
                <h4>Gridadmin file</h4>
                <p>HDF5-file (*.h5) containing a 3Di computational grid. Note that gridadmin files generated on the server contain exchange levels for 2D flowlines, whereas locally generated gridadmin files do not. In the latter case, the processing algorithm will analyse the DEM to obtain these values.</p>
                <h4>Digital elevation model</h4>
                <p>Raster of the schematisation's digital elevation model (DEM).</p>
                <h4>Linear obstacles</h4>
                <p>Obstacles in this layer will be used to update cell edge exchange levels, <i>in addition to</i> any obstacles already present in the gridadmin file (i.e. in files that were downloaded from the server). This input must be a vector layer with line geometry and a <i>crest_level</i> field</p>
                <h4>Flowlines</h4>
                <p>Can be used to limit the analysis to a specific part of the computational grid. For example, select flowlines that have a total discharge of > 10 m<sup>3</sup></p>
                <h4>Minimum obstacle height (m)</h4>
                <p>Only obstacles with a crest level that is significantly higher than the exchange level will be identified. 'Significantly higher' is defined as <em>crest level &gt; exchange level + minimum obstacle height</em>.</p>
                <h4>Vertical search precision (m)</h4>
                <p>The crest level of the identified obstacle will always be within <em>vertical search precision</em> of the actual crest level. A smaller value will yield more precise results; a higher value will make the algorithm faster to execute.</p>
                <h3>Outputs</h3>
                <h4>Obstacle in DEM&nbsp;</h4>
                <p>Approximate location of the obstacle in the DEM. Its geometry is a straight line between the highest pixels of the obstacle on the cell edges. Attributes:</p>
                <ul>
                <li> id: id of the flowline this obstacle affects. If more than one flowline is affected, the flowline with the lowest exchange level is used. </li>
                <li> crest_level: lowest elevation at which water can flow over the obstacle </li>
                <li> exchange_level: lowest exchange level of the cell edge(s) that this obstacle applies to </li>
                </ul>
                <p>The styling is shows the difference between the crest level and the exchange level</p>
                <h4>Obstacle on cell edge</h4>
                <p>Suggested obstacle to add to the schematisation. In most cases, it is recommended solve any leaking obstacle issues with grid refinement, and only add obstacles if this does not solve the issue.</p>
                <p>The styling shows the difference between the crest level and the exchange level</p>
            """

    def createInstance(self):
        return DetectLeakingObstaclesAlgorithm()


class DetectLeakingObstaclesWithDischargeThresholdAlgorithm(DetectLeakingObstaclesAlgorithm):
    INPUT_RESULTS_THREEDI = "INPUT_RESULTS_THREEDI"
    INPUT_MIN_DISCHARGE = "INPUT_MIN_DISCHARGE"

    def initAlgorithm(self, config):
        super().initAlgorithm(config)
        self.addParameter(
            QgsProcessingParameterFile(
                self.INPUT_RESULTS_THREEDI, "3Di Results file", extension="nc"
            )
        )

        min_discharge_param = QgsProcessingParameterNumber(
            self.INPUT_MIN_DISCHARGE,
            "Minimum cumulative discharge (m3)",
            type=QgsProcessingParameterNumber.Type.Double
        )
        min_discharge_param.setMetadata({"widget_wrapper": {"decimals": 3}})
        self.addParameter(min_discharge_param)

    def sink_field_data(self):
        return super().sink_field_data() + [
            {"name": "discharge_without_obstacle", "type": QVariant.Double},
            {"name": "discharge_with_obstacle", "type": QVariant.Double},
            {"name": "discharge_reduction", "type": QVariant.Double}
        ]

    def read_parameters(self, parameters, context, feedback):
        super().read_parameters(parameters, context, feedback)
        self.min_discharge = self.parameterAsDouble(parameters, self.INPUT_MIN_DISCHARGE, context)
        self.results_threedi_fn = self.parameterAsFile(parameters, self.INPUT_RESULTS_THREEDI, context)
        self.grid_result_admin = GridH5ResultAdmin(self.gridadmin_fn, self.results_threedi_fn)

    def get_leak_detector(self, feedback):
        leak_detector = LeakDetectorWithDischargeThreshold(
            grid_result_admin=self.grid_result_admin,
            dem=self.dem_ds,
            flowline_ids=self.flowline_ids,
            min_obstacle_height=self.min_obstacle_height,
            min_discharge=self.min_discharge,
            obstacles=self.input_obstacles,
            feedback=feedback
        )
        return leak_detector

    def name(self):
        """
        Returns the algorithm name, used for identifying the algorithm
        """
        return "detect_leaking_obstacles_q_threshold"

    def displayName(self):
        """
        Returns the algorithm name, which should be used for any
        user-visible display of the algorithm name.
        """
        return "Detect leaking obstacles in DEM (discharge threshold)"

    def createInstance(self):
        return DetectLeakingObstaclesWithDischargeThresholdAlgorithm()

1	# -- coding: utf-8 --
2
3	"""	1✔
4	***************************************************************************
5	* *
6	* This program is free software; you can redistribute it and/or modify *
7	* it under the terms of the GNU General Public License as published by *
8	* the Free Software Foundation; either version 2 of the License, or *
9	* (at your option) any later version. *
10	* *
11	***************************************************************************
12	"""
13	from pathlib import Path	1✔
14	from osgeo import gdal	1✔
15	from typing import Any, Dict, Tuple, Iterator, List	1✔
16	from shapely import wkt	1✔
17	from threedigrid.admin.gridadmin import GridH5Admin	1✔
18	from threedigrid.admin.gridresultadmin import GridH5ResultAdmin	1✔
19	from qgis.PyQt.QtCore import QVariant	1✔
20	from qgis.core import (	1✔
21	QgsCoordinateReferenceSystem,
22	QgsGeometry,
23	QgsFeature,
24	QgsFeatureSink,
25	QgsField,
26	QgsFields,
27	QgsProcessing,
28	QgsProcessingAlgorithm,
29	QgsProcessingContext,
30	QgsProcessingException,
31	QgsProcessingParameterFeatureSink,
32	QgsProcessingParameterFeatureSource,
33	QgsProcessingParameterFile,
34	QgsProcessingParameterNumber,
35	QgsProcessingParameterRasterLayer,
36	QgsWkbTypes,
37	)
38
39	from threedi_results_analysis.processing.deps.discharge.leak_detector import LeakDetector	1✔
40	from threedi_results_analysis.processing.deps.discharge.discharge_reduction import LeakDetectorWithDischargeThreshold	1✔
41	from threedi_results_analysis.utils.threedi_result_aggregation.aggregation_classes import Aggregation, AggregationSign	1✔
42	from threedi_results_analysis.utils.threedi_result_aggregation.constants import AGGREGATION_VARIABLES, AGGREGATION_METHODS	1✔
43
44	Q_NET_SUM = Aggregation(	1✔
45	variable=AGGREGATION_VARIABLES.get_by_short_name("q"),
46	method=AGGREGATION_METHODS.get_by_short_name("sum"),
47	sign=AggregationSign("net", "Net"),
48	)
49
50	STYLE_DIR = Path(__file__).parent / "styles"	1✔
51
52	QVARIANT_PYTHON_TYPES = {	1✔
53	QVariant.Int: int,
54	QVariant.Double: float
55	}
56
57
58	class DetectLeakingObstaclesBase(QgsProcessingAlgorithm):	1✔
59	"""
60	Base algorithm for 'Detect leaking obstacles' algorithms, not to be exposed to users
61	"""
62
63	INPUT_GRIDADMIN = "INPUT_GRIDADMIN"	1✔
64	INPUT_DEM = "INPUT_DEM"	1✔
65	INPUT_FLOWLINES = "INPUT_FLOWLINES"	1✔
66	INPUT_OBSTACLES = "INPUT_OBSTACLES"	1✔
67	INPUT_MIN_OBSTACLE_HEIGHT = "INPUT_MIN_OBSTACLE_HEIGHT"	1✔
68
69	OUTPUT_EDGES = "OUTPUT_EDGES"	1✔
70	OUTPUT_OBSTACLES = "OUTPUT_OBSTACLES"	1✔
71
72	def initAlgorithm(self, config):	1✔
73
74	self.addParameter(	×
75	QgsProcessingParameterFile(
76	self.INPUT_GRIDADMIN, "Gridadmin file", extension="h5"
77	)
78	)
79
80	self.addParameter(	×
81	QgsProcessingParameterRasterLayer(
82	self.INPUT_DEM, "Digital Elevation Model"
83	)
84	)
85
86	self.addParameter(	×
87	QgsProcessingParameterFeatureSource(
88	self.INPUT_OBSTACLES,
89	"Linear obstacles",
90	[QgsProcessing.SourceType.TypeVectorLine],
91	optional=True
92	)
93	)
94
95	self.addParameter(	×
96	QgsProcessingParameterFeatureSource(
97	self.INPUT_FLOWLINES,
98	"Flowlines",
99	[QgsProcessing.SourceType.TypeVectorLine],
100	optional=True
101	)
102	)
103
104	min_obstacle_height_param = QgsProcessingParameterNumber(	×
105	self.INPUT_MIN_OBSTACLE_HEIGHT,
106	"Minimum obstacle height (m)",
107	type=QgsProcessingParameterNumber.Type.Double
108	)
109	min_obstacle_height_param.setMetadata({"widget_wrapper": {"decimals": 3}})	×
110	self.addParameter(min_obstacle_height_param)	×
111
112	self.addParameter(	×
113	QgsProcessingParameterFeatureSink(
114	self.OUTPUT_EDGES,
115	"Output: Obstacle on cell edge",
116	type=QgsProcessing.SourceType.TypeVectorLine
117	)
118	)
119
120	self.addParameter(	×
121	QgsProcessingParameterFeatureSink(
122	self.OUTPUT_OBSTACLES,
123	"Output: Obstacle in DEM",
124	type=QgsProcessing.SourceType.TypeVectorLine
125	)
126	)
127
128	def checkParameterValues(self, parameters: Dict[str, Any], context: QgsProcessingContext) -> Tuple[bool, str]:	1✔
129	success, msg = super().checkParameterValues(parameters, context)	×
130	if success:	×
131	msg_list = list()	×
132
133	# check if min_obstacle_height > 0
134	min_obstacle_height = self.parameterAsDouble(parameters, self.INPUT_MIN_OBSTACLE_HEIGHT, context)	×
135	if min_obstacle_height <= 0:	×
136	msg_list.append('Minimum obstacle height must be greater than 0')	×
137
138	# check input obstacles has a field called crest_level
139	input_obstacles_source = self.parameterAsSource(parameters, self.INPUT_OBSTACLES, context)	×
140	if input_obstacles_source:	×
141	if 'crest_level' not in input_obstacles_source.fields().names():	×
142	msg_list.append('Obstacle lines layer does not contain crest_level field')	×
143	success = len(msg_list) == 0	×
144	msg = '; '.join(msg_list)	×
145	return success, msg	×
146
147	@staticmethod	1✔
148	def sink_field_data() -> List[Dict]:	1✔
149	"""
150	Return a list of dicts that contain the data needed to defined fields in the feature sinks (output layers).
151	Each dict contains the name (str) and the type (QVariant)
152	"""
153	return [	×
154	{"name": "flowline_id", "type": QVariant.Int},
155	{"name": "exchange_level", "type": QVariant.Double},
156	{"name": "crest_level", "type": QVariant.Double}
157	]
158
159	def add_features_to_sink(self, feedback, sink: QgsFeatureSink, features_data: Iterator):	1✔
160	for i, feature_data in enumerate(features_data):	×
161	if feedback.isCanceled():	×
162	return {}	×
163	feature = QgsFeature()	×
164	feature.setFields(self.sink_fields)	×
165	for i, field in enumerate(self.sink_field_data()):	×
166	convert = QVARIANT_PYTHON_TYPES[field["type"]]	×
167	feature.setAttribute(i, convert(feature_data[field["name"]]))	×
168	geometry = QgsGeometry()	×
169	geometry.fromWkb(feature_data["geometry"].wkb)	×
170	feature.setGeometry(geometry)	×
NEW 171	sink.addFeature(feature, QgsFeatureSink.Flag.FastInsert)	×
172
173	def read_parameters(self, parameters, context, feedback):	1✔
174	self.gridadmin_fn = self.parameterAsFile(parameters, self.INPUT_GRIDADMIN, context)	×
175	self.gridadmin = GridH5Admin(self.gridadmin_fn)	×
176	dem = self.parameterAsRasterLayer(parameters, self.INPUT_DEM, context)	×
177	dem_fn = dem.dataProvider().dataSourceUri()	×
178	self.dem_ds = gdal.Open(dem_fn)	×
179	flowlines_source = self.parameterAsSource(parameters, self.INPUT_FLOWLINES, context)	×
180	obstacles_source = self.parameterAsSource(parameters, self.INPUT_OBSTACLES, context)	×
181	self.min_obstacle_height = self.parameterAsDouble(parameters, self.INPUT_MIN_OBSTACLE_HEIGHT, context)	×
182
183	crs = QgsCoordinateReferenceSystem(f"EPSG:{self.gridadmin.epsg_code}")	×
184
185	self.sink_fields = QgsFields()	×
186	for field_data in self.sink_field_data():	×
187	self.sink_fields.append(QgsField(**field_data))	×
188
189	self.edges_sink, self.edges_sink_dest_id = self.parameterAsSink(	×
190	parameters,
191	self.OUTPUT_EDGES,
192	context,
193	fields=self.sink_fields,
194	geometryType=QgsWkbTypes.Type.LineString,
195	crs=crs
196	)
197	self.obstacles_sink, self.obstacles_sink_dest_id = self.parameterAsSink(	×
198	parameters,
199	self.OUTPUT_OBSTACLES,
200	context,
201	fields=self.sink_fields,
202	geometryType=QgsWkbTypes.Type.LineString,
203	crs=crs
204	)
205
206	# get list of flowline ids
207	if flowlines_source:	×
208	field_index = flowlines_source.fields().indexFromName('id')	×
209	self.flowline_ids = [feature.attributes()[field_index] for feature in flowlines_source.getFeatures()]	×
210	else:
211	self.flowline_ids = list(self.gridadmin.lines.id)	×
212
213	if obstacles_source:	×
214	feedback.setProgressText("Read linear obstacles input...")	×
215	if obstacles_source.sourceCrs() != crs:	×
216	raise QgsProcessingException(	×
217	"Obstacles input has different Coordinate Reference System than the gridadmin file"
218	)
219	crest_level_field_idx = obstacles_source.fields().indexFromName("crest_level")	×
220
221	self.input_obstacles = list()	×
222	for input_obstacle in obstacles_source.getFeatures():	×
223	geom = wkt.loads(input_obstacle.geometry().asWkt())	×
224	crest_level = float(input_obstacle[crest_level_field_idx])	×
225	self.input_obstacles.append((geom, crest_level))	×
226	else:
227	self.input_obstacles = None	×
228
229	def get_leak_detector(self, feedback):	1✔
230	leak_detector = LeakDetector(	×
231	gridadmin=self.gridadmin,
232	dem=self.dem_ds,
233	flowline_ids=self.flowline_ids,
234	min_obstacle_height=self.min_obstacle_height,
235	obstacles=self.input_obstacles,
236	feedback=feedback
237	)
238	return leak_detector	×
239
240	def processAlgorithm(self, parameters, context, feedback):	1✔
241	self.read_parameters(parameters, context, feedback)	×
242	feedback.setProgressText("Read computational grid...")	×
243	leak_detector = self.get_leak_detector(feedback)	×
244	if feedback.isCanceled():	×
245	return {}	×
246	feedback.setProgressText("Find obstacles...")	×
247	leak_detector.run(feedback=feedback)	×
248	feedback.setProgressText("Create 'Obstacle on cell edge' features...")	×
249	self.add_features_to_sink(	×
250	feedback=feedback,
251	sink=self.edges_sink,
252	features_data=leak_detector.results(geometry='EDGE')
253	)
254	feedback.setProgressText("Create 'Obstacle in DEM' features...")	×
255	self.add_features_to_sink(	×
256	feedback=feedback,
257	sink=self.obstacles_sink,
258	features_data=leak_detector.results(geometry='OBSTACLE')
259	)
260
261	return {	×
262	self.OUTPUT_EDGES: self.edges_sink_dest_id,
263	self.OUTPUT_OBSTACLES: self.obstacles_sink_dest_id
264	}
265
266	def group(self):	1✔
267	return "Computational Grid"	×
268
269	def groupId(self):	1✔
270	return "computational_grid"	×
271
272
273	class DetectLeakingObstaclesAlgorithm(DetectLeakingObstaclesBase):	1✔
274	"""
275	Detect obstacle lines in the DEM that are ignored by 3Di due to its location relative to cell edges
276	"""
277	def postProcessAlgorithm(self, context, feedback):	1✔
278	"""Set styling of output vector layers"""
279	edges_layer = context.getMapLayer(self.edges_sink_dest_id)	×
280	edges_layer.loadNamedStyle(str(STYLE_DIR / "obstacle_on_cell_edge.qml"))	×
281
282	obstacles_layer = context.getMapLayer(self.obstacles_sink_dest_id)	×
283	obstacles_layer.loadNamedStyle(str(STYLE_DIR / "obstacle_in_dem.qml"))	×
284
285	return {	×
286	self.OUTPUT_EDGES: self.edges_sink_dest_id,
287	self.OUTPUT_OBSTACLES: self.obstacles_sink_dest_id
288	}
289
290	def name(self):	1✔
291	"""
292	Returns the algorithm name, used for identifying the algorithm
293	"""
294	return "detect_leaking_obstacles"	×
295
296	def displayName(self):	1✔
297	"""
298	Returns the algorithm name, which should be used for any
299	user-visible display of the algorithm name.
300	"""
301	return "Detect leaking obstacles in DEM"	×
302
303	def shortHelpString(self):	1✔
304	return """	×
305	<h3>Introduction</h3>
306	<p>The elevation at which flow between 2D cells is possible (the 'exchange level'), depends on the elevation of the pixels directly adjacent to the cell edge. Obstacles in the DEM that do not cover the entire edge will therefore not stop the flow, i.e. water 'leaks' through the obstacle. This is more likely to occur if obstacles are diagonal and/or narrow compared to the computational grid size.</p>
307	<p>This processing algorithm detects such cases. Please inspect the locations where the algorithm identifies leaking obstacles and add grid refinement and/or obstacles to the schematisation to solve the issue if needed.</p>
308	<h3>Parameters</h3>
309	<h4>Gridadmin file</h4>
310	<p>HDF5-file (*.h5) containing a 3Di computational grid. Note that gridadmin files generated on the server contain exchange levels for 2D flowlines, whereas locally generated gridadmin files do not. In the latter case, the processing algorithm will analyse the DEM to obtain these values.</p>
311	<h4>Digital elevation model</h4>
312	<p>Raster of the schematisation's digital elevation model (DEM).</p>
313	<h4>Linear obstacles</h4>
314	<p>Obstacles in this layer will be used to update cell edge exchange levels, <i>in addition to</i> any obstacles already present in the gridadmin file (i.e. in files that were downloaded from the server). This input must be a vector layer with line geometry and a <i>crest_level</i> field</p>
315	<h4>Flowlines</h4>
316	<p>Can be used to limit the analysis to a specific part of the computational grid. For example, select flowlines that have a total discharge of > 10 m<sup>3</sup></p>
317	<h4>Minimum obstacle height (m)</h4>
318	<p>Only obstacles with a crest level that is significantly higher than the exchange level will be identified. 'Significantly higher' is defined as <em>crest level > exchange level + minimum obstacle height</em>.</p>
319	<h4>Vertical search precision (m)</h4>
320	<p>The crest level of the identified obstacle will always be within <em>vertical search precision</em> of the actual crest level. A smaller value will yield more precise results; a higher value will make the algorithm faster to execute.</p>
321	<h3>Outputs</h3>
322	<h4>Obstacle in DEM </h4>
323	<p>Approximate location of the obstacle in the DEM. Its geometry is a straight line between the highest pixels of the obstacle on the cell edges. Attributes:</p>
324	<ul>
325	<li> id: id of the flowline this obstacle affects. If more than one flowline is affected, the flowline with the lowest exchange level is used. </li>
326	<li> crest_level: lowest elevation at which water can flow over the obstacle </li>
327	<li> exchange_level: lowest exchange level of the cell edge(s) that this obstacle applies to </li>
328	</ul>
329	<p>The styling is shows the difference between the crest level and the exchange level</p>
330	<h4>Obstacle on cell edge</h4>
331	<p>Suggested obstacle to add to the schematisation. In most cases, it is recommended solve any leaking obstacle issues with grid refinement, and only add obstacles if this does not solve the issue.</p>
332	<p>The styling shows the difference between the crest level and the exchange level</p>
333	"""
334
335	def createInstance(self):	1✔
336	return DetectLeakingObstaclesAlgorithm()	×
337
338
339	class DetectLeakingObstaclesWithDischargeThresholdAlgorithm(DetectLeakingObstaclesAlgorithm):	1✔
340	INPUT_RESULTS_THREEDI = "INPUT_RESULTS_THREEDI"	1✔
341	INPUT_MIN_DISCHARGE = "INPUT_MIN_DISCHARGE"	1✔
342
343	def initAlgorithm(self, config):	1✔
344	super().initAlgorithm(config)	×
345	self.addParameter(	×
346	QgsProcessingParameterFile(
347	self.INPUT_RESULTS_THREEDI, "3Di Results file", extension="nc"
348	)
349	)
350
351	min_discharge_param = QgsProcessingParameterNumber(	×
352	self.INPUT_MIN_DISCHARGE,
353	"Minimum cumulative discharge (m3)",
354	type=QgsProcessingParameterNumber.Type.Double
355	)
356	min_discharge_param.setMetadata({"widget_wrapper": {"decimals": 3}})	×
357	self.addParameter(min_discharge_param)	×
358
359	def sink_field_data(self):	1✔
360	return super().sink_field_data() + [	×
361	{"name": "discharge_without_obstacle", "type": QVariant.Double},
362	{"name": "discharge_with_obstacle", "type": QVariant.Double},
363	{"name": "discharge_reduction", "type": QVariant.Double}
364	]
365
366	def read_parameters(self, parameters, context, feedback):	1✔
367	super().read_parameters(parameters, context, feedback)	×
368	self.min_discharge = self.parameterAsDouble(parameters, self.INPUT_MIN_DISCHARGE, context)	×
369	self.results_threedi_fn = self.parameterAsFile(parameters, self.INPUT_RESULTS_THREEDI, context)	×
370	self.grid_result_admin = GridH5ResultAdmin(self.gridadmin_fn, self.results_threedi_fn)	×
371
372	def get_leak_detector(self, feedback):	1✔
373	leak_detector = LeakDetectorWithDischargeThreshold(	×
374	grid_result_admin=self.grid_result_admin,
375	dem=self.dem_ds,
376	flowline_ids=self.flowline_ids,
377	min_obstacle_height=self.min_obstacle_height,
378	min_discharge=self.min_discharge,
379	obstacles=self.input_obstacles,
380	feedback=feedback
381	)
382	return leak_detector	×
383
384	def name(self):	1✔
385	"""
386	Returns the algorithm name, used for identifying the algorithm
387	"""
388	return "detect_leaking_obstacles_q_threshold"	×
389
390	def displayName(self):	1✔
391	"""
392	Returns the algorithm name, which should be used for any
393	user-visible display of the algorithm name.
394	"""
395	return "Detect leaking obstacles in DEM (discharge threshold)"	×
396
397	def createInstance(self):	1✔
398	return DetectLeakingObstaclesWithDischargeThresholdAlgorithm()	×

nens / ThreeDiToolbox / #2600

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous