#2589

Committed 19 Sep 2025 08:50AM UTC coverage: 35.01% (-0.1%) from 35.146%

Build # #2589

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge 38792c162 into f6f4be1e7

Coverage Stats

62 of 260 new or added lines in 40 files covered. (23.85%)

6 existing lines in 5 files now uncovered.

4859 of 13879 relevant lines covered (35.01%)

0.35 hits per line

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

23.53

/threedi_plugin_layer_manager.py

import os
from collections import OrderedDict
from pathlib import Path
import re
import uuid
from qgis.PyQt.QtCore import Qt
from threedigrid.admin.exporters.geopackage import GeopackageExporter
from qgis.PyQt.QtCore import QObject, pyqtSlot, pyqtSignal, QVariant
from qgis.core import QgsVectorLayer, QgsProject, QgsMapLayer, QgsField, QgsWkbTypes, QgsLayerTreeNode
from threedi_results_analysis.threedi_plugin_model import ThreeDiGridItem, ThreeDiResultItem
from threedi_results_analysis.utils.constants import TOOLBOX_QGIS_GROUP_NAME, TOOLBOX_MESSAGE_TITLE
from threedi_results_analysis.utils.user_messages import StatusProgressBar, messagebar_message, pop_up_critical
from threedi_results_analysis.utils.utils import safe_join
from qgis.utils import iface

styles_dir = os.path.join(os.path.dirname(os.path.realpath(__file__)), "layer_styles", "grid")

import logging
logger = logging.getLogger(__name__)

GRID_GROUP_NAME = "Computational grid"


def dirty(func):
    """
    This decorator ensures the QGIS project is marked as dirty when
    the function is done.
    """
    def wrapper(*args, **kwargs):
        func(*args, **kwargs)
        QgsProject.instance().setDirty()
    return wrapper


def copy_layer_into_memory_layer(source_layer, layer_name, dest_layer):

    source_provider = source_layer.dataProvider()

    uri = "{0}?crs=EPSG:{1}".format(
        QgsWkbTypes.displayString(source_provider.wkbType()).lstrip("WKB"),
        str(source_provider.crs().postgisSrid()),
    )

    if dest_layer is None:
        dest_layer = QgsVectorLayer(uri, layer_name, "memory")
    else:
        logger.info("Reusing memory layer instance")

    dest_provider = dest_layer.dataProvider()

    if not dest_provider.addAttributes(source_provider.fields()):
        logger.error(dest_provider.lastError())
    dest_layer.updateFields()

    if not dest_provider.addFeatures(source_provider.getFeatures()):
        logger.error(dest_provider.lastError())
    dest_layer.updateExtents()

    if source_provider.featureCount() != dest_provider.featureCount():
        messagebar_message("Error", "Not all features are loaded in layer {layer_name}!", level=2, duration=5)

    return dest_layer


# Layers need to be in specific order and naming:
gpkg_layers = OrderedDict(
    [
        ("Model properties", "meta"),
        ("Pump (point)", "pump"),
        ("Node", "node"),
        ("Pump (line)", "pump_linestring"),
        ("Flowline", "flowline"),
        ("Cell", "cell"),
        ("Obstacle", "obstacle"),
    ]
)


class ThreeDiPluginLayerManager(QObject):
    """
    The Layer manager creates layers from a geopackage and keeps track
    of the connection between model items (grids) and layers.

    In case a model item is deleted, the corresponding layers are also
    deleted.
    """
    grid_loaded = pyqtSignal(ThreeDiGridItem)
    result_loaded = pyqtSignal(ThreeDiResultItem, ThreeDiGridItem)
    grid_unloaded = pyqtSignal(ThreeDiGridItem)
    result_unloaded = pyqtSignal(ThreeDiResultItem)

    # error signals so the UI can deal with this accordingly
    grid_not_loaded = pyqtSignal(ThreeDiGridItem)
    result_not_loaded = pyqtSignal(ThreeDiResultItem, ThreeDiGridItem)

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    @pyqtSlot(ThreeDiGridItem)
    def load_grid(self, grid_item: ThreeDiGridItem) -> bool:
        # generate geopackage if needed and point item path to it
        if grid_item.path.suffix == ".h5":
            path_h5 = grid_item.path
            path_gpkg = path_h5.with_suffix(".gpkg")
            if not path_gpkg.exists():
                self.__class__._generate_gpkg(path_h5=path_h5, path_gpkg=path_gpkg)
            grid_item.path = path_gpkg
        else:
            path_gpkg = grid_item.path

        # Text of item is determined by folder structure
        if not grid_item.text():
            grid_item.setText(ThreeDiPluginLayerManager._resolve_grid_item_text(grid_item.path))

        if not ThreeDiPluginLayerManager._add_layers_from_gpkg(path_gpkg, grid_item):
            pop_up_critical("Failed adding the layers to the project.")
            self.grid_not_loaded.emit(grid_item)
            return False

        messagebar_message(TOOLBOX_MESSAGE_TITLE, "Added layers to the project", duration=2)

        self.grid_loaded.emit(grid_item)
        return True

    @pyqtSlot(ThreeDiGridItem)
    def unload_grid(self, item: ThreeDiGridItem) -> bool:
        """Removes the corresponding layers from the group in the project"""

        # It could be possible that some layers have been dragged outside the
        # layer group. Delete the individual layers first
        for layer_id in item.layer_ids.values():
            assert QgsProject.instance().mapLayer(layer_id)
            QgsProject.instance().removeMapLayer(layer_id)
        item.layer_ids.clear()

        # Deletion of root node of a tree will delete all nodes of the tree.
        # In case the user dragged another layer in the group, remove the reference
        # from this grid to the group, but don't delete it from QGIS.
        assert item.layer_group
        grid_group = item.layer_group.findGroup(GRID_GROUP_NAME)

        # Remove "Computational Grid" group
        if len(grid_group.children()) == 0:
            item.layer_group.removeChildNode(grid_group)

        if len(item.layer_group.children()) == 0:
            item.layer_group.parent().removeChildNode(item.layer_group)
        else:
            logger.info(f"Item group of grid {item.text()} contains external layers: not removing.")

        item.layer_group = None
        iface.mapCanvas().refresh()

        self.grid_unloaded.emit(item)

    @dirty
    @pyqtSlot(ThreeDiGridItem)
    def update_grid(self, item: ThreeDiGridItem) -> bool:
        """Updates the group name in the project"""
        assert item.layer_group
        item.layer_group.setName(item.text())
        return True

    @pyqtSlot(ThreeDiResultItem, ThreeDiGridItem)
    def load_result(self, threedi_result_item: ThreeDiResultItem, grid_item: ThreeDiGridItem) -> bool:

        # Text of item is determined by folder structure
        if not threedi_result_item.text():
            threedi_result_item.setText(ThreeDiPluginLayerManager._resolve_result_item_text(threedi_result_item.path))

        # Add result fields for this result to the grid layers
        logger.info("Adding result fields to grid layers")
        for layer_id in grid_item.layer_ids.values():
            layer = QgsProject.instance().mapLayer(layer_id)
            provider = layer.dataProvider()

            # Generate a random field name, with the result text
            # as alias (display) name.
            unique_identifier = str(uuid.uuid4())

            result_field_name = "result_" + unique_identifier
            result_field = QgsField(result_field_name, QVariant.Double)
            result_field.setAlias(threedi_result_item.text())

            initial_value_field_name = "initial_value_" + unique_identifier
            initial_value_field = QgsField(initial_value_field_name, QVariant.Double)
            initial_value_field.setAlias(threedi_result_item.text() + "_initial_value")

            # Check for duplicate field names (even though QGIS does not allow
            # addition of QgsFields (both attribute or expression) with already
            # existing names AND the generated layers are marked READONLY)
            if (layer.fields().indexFromName(result_field_name) != -1 or
                    layer.fields().indexFromName(initial_value_field_name) != -1):
                logger.error("Field already exist, aborting addition.")
                self.result_not_loaded.emit(threedi_result_item, grid_item)
                return False

            provider.addAttributes([result_field, initial_value_field])
            layer.updateFields()

            # Store the added field names so we can remove the field when the result is removed
            threedi_result_item._result_field_names[layer_id] = (result_field_name, initial_value_field_name)

        self.result_loaded.emit(threedi_result_item, grid_item)
        return True

    @pyqtSlot(ThreeDiResultItem)
    def unload_result(self, threedi_result_item: ThreeDiResultItem) -> bool:
        # Remove the corresponding result fields from the grid layers
        for layer_id, result_field_names in threedi_result_item._result_field_names.items():
            # It could be that the map layer is removed by QGIS
            if QgsProject.instance().mapLayer(layer_id) is not None:
                layer = QgsProject.instance().mapLayer(layer_id)
                provider = layer.dataProvider()

                assert len(result_field_names) == 2
                idx = layer.fields().indexFromName(result_field_names[0])
                assert idx != -1
                provider.deleteAttributes([idx])
                layer.updateFields()

                idx = layer.fields().indexFromName(result_field_names[1])
                assert idx != -1
                provider.deleteAttributes([idx])
                layer.updateFields()

        threedi_result_item._result_field_names.clear()

        # In case the corresponding grid contains no more (selected) results, the
        # original styling should be reset
        reset_styling = True
        grid_item = threedi_result_item.parent()
        assert isinstance(grid_item, ThreeDiGridItem)
        if grid_item.hasChildren():
            for i in range(grid_item.rowCount()):
                result_item = grid_item.child(i)
                if (result_item.checkState() == Qt.CheckState.Checked and threedi_result_item is not result_item):
                    reset_styling = False

        if reset_styling:
            self.reset_styling(grid_item)

        self.result_unloaded.emit(threedi_result_item)
        return True

    @dirty
    @pyqtSlot(ThreeDiResultItem)
    def result_unchecked(self, item: ThreeDiResultItem):
        # In case all results are unchecked, revert back to default styling (and naming)
        grid_item = item.parent()
        assert isinstance(grid_item, ThreeDiGridItem)
        if grid_item.hasChildren():
            for i in range(grid_item.rowCount()):
                result_item = grid_item.child(i)
                if result_item.checkState() == Qt.CheckState.Checked:
                    return

        self.reset_styling(grid_item)

    @pyqtSlot(ThreeDiGridItem)
    def reset_styling(self, grid_item: ThreeDiGridItem) -> None:
        """Sets all the grid layers for a given grid back to their original name and style"""
        for layer_name, table_name in gpkg_layers.items():

            # Some models do not contain pump or obstacle layers.
            if table_name not in grid_item.layer_ids.keys():
                continue

            scratch_layer = QgsProject.instance().mapLayer(grid_item.layer_ids[table_name])
            assert scratch_layer

            # (Re)apply the style and naming
            qml_path = safe_join(styles_dir, f"{table_name}.qml")
            if os.path.exists(qml_path):
                msg, res = scratch_layer.loadNamedStyle(qml_path)
                if not res:
                    logger.error(f"Unable to load style: {msg}")

            scratch_layer.setName(layer_name)
            iface.layerTreeView().refreshLayerSymbology(scratch_layer.id())
            scratch_layer.triggerRepaint()

    @dirty
    @pyqtSlot(ThreeDiResultItem)
    def update_result(self, threedi_result_item: ThreeDiResultItem) -> bool:
        # Update the display name of the result fields
        logger.info("Updating result fields")
        for layer_id, result_field_names in threedi_result_item._result_field_names.items():
            layer = QgsProject.instance().mapLayer(layer_id)

            assert len(result_field_names) == 2
            idx = layer.fields().indexFromName(result_field_names[0])
            assert idx != -1
            layer.setFieldAlias(idx, threedi_result_item.text())

            idx = layer.fields().indexFromName(result_field_names[1])
            assert idx != -1
            layer.setFieldAlias(idx, threedi_result_item.text() + '_initial_value')

        return True

    @staticmethod
    def _generate_gpkg(path_h5, path_gpkg) -> None:
        progress_bar = StatusProgressBar(100, "Generating computational grid geopackage")
        exporter = GeopackageExporter(path_h5.open('rb'), str(path_gpkg))
        exporter.export(
            lambda count, total, pb=progress_bar: pb.set_value((count * 100) // total)
        )
        del progress_bar

        messagebar_message(TOOLBOX_MESSAGE_TITLE, "Generated computational grid geopackage")

    @staticmethod
    def _add_layers_from_gpkg(path, item: ThreeDiGridItem) -> bool:
        """
        Retrieves (a subset of the) layers from gpk and add to project.
        """

        invalid_layers = []
        empty_layers = []

        item.layer_group = ThreeDiPluginLayerManager._get_or_create_group(item.text())

        # Use to modify grid name when LayerGroup is renamed
        item.layer_group.nameChanged.connect(lambda node, txt, grid_item=item: ThreeDiPluginLayerManager._layer_node_renamed(node, txt, grid_item))

        progress_bar = StatusProgressBar(len(gpkg_layers) - 1, "Adding computational grid layers")
        for layer_name, table_name in gpkg_layers.items():

            # QGIS does save memory layers to the project file (but without the data)
            # Removing the scratch layer and resaving the project causes QGIS to crash,
            # therefore we reuse the layer instance.
            scratch_layer = None
            if table_name in item.layer_ids.keys():
                scratch_layer = QgsProject.instance().mapLayer(item.layer_ids[table_name])
                if scratch_layer:
                    logger.info(f"Map layer corresponding to table {item.layer_ids[table_name]} already exist in project, reusing...")

            # Using the QgsInterface function addVectorLayer shows (annoying) confirmation dialogs
            # iface.addVectorLayer(gpkg_file + "|layername=" + layer, layer, 'ogr')
            vector_layer = QgsVectorLayer(str(path) + "|layername=" + table_name, layer_name, "ogr")
            if not vector_layer.isValid():
                invalid_layers.append(layer_name)
                continue

            # Only load layers that contain some features
            if not vector_layer.featureCount():
                empty_layers.append(layer_name)
                continue

            vector_layer = copy_layer_into_memory_layer(
                vector_layer, layer_name, scratch_layer
            )

            # Apply the style
            qml_path = safe_join(styles_dir, f"{table_name}.qml")
            if os.path.exists(qml_path):
                msg, res = vector_layer.loadNamedStyle(qml_path)
                if not res:
                    logger.error(f"Unable to load style: {msg}")
                # prior to QGIS 3.24, saveStyleToDatabase would show an (annoying) message box
                # warning when a style with the same styleName already existed. Unfortunately,
                # QgsProviderRegistry::styleExists is not available in Python
                # if table_name not in vector_layer.listStylesInDatabase()[2]:
                    # Memory providers do not support saving of styles, commented
                    # msg = vector_layer.saveStyleToDatabase(table_name, "", True, "")
                    # if msg:
                    #    logger.error(f"Unable to save style to DB: {msg}")

            if not vector_layer.setReadOnly(True):
                logger.error(f"Unable to set layer {table_name} to read-only")
            vector_layer.setFlags(QgsMapLayer.LayerFlag.Searchable | QgsMapLayer.LayerFlag.Identifiable)
            # Require to keep track of the purpose of the layer (node, pump, flowline etc)
            # Can also be used to check whether this layer is generated by the Toolbox
            vector_layer.setObjectName(table_name)

            if scratch_layer is None:
                # Keep track of layer id for future reference (deletion of grid item)
                item.layer_ids[table_name] = vector_layer.id()

                QgsProject.instance().addMapLayer(vector_layer, addToLegend=False)
                # Add to computational grid subgroup (created above)
                item.layer_group.findGroup(GRID_GROUP_NAME).addLayer(vector_layer)

            progress_bar.increase_progress()

        del progress_bar

        # Invalid layers info
        if invalid_layers:
            logger.warning("The following layers are missing or invalid:\n * " + "\n * ".join(invalid_layers) + "\n\n")

        # Empty layers info
        if empty_layers:
            logger.warning("The following layers contained no feature:\n * " + "\n * ".join(empty_layers) + "\n\n")

        return True

    @staticmethod
    def _get_or_create_group(group_name: str):
        root = QgsProject.instance().layerTreeRoot()
        root_group = root.findGroup(TOOLBOX_QGIS_GROUP_NAME)
        if not root_group:
            root_group = root.insertGroup(0, TOOLBOX_QGIS_GROUP_NAME)

        layer_group = root_group.findGroup(group_name)
        if not layer_group:
            layer_group = root_group.insertGroup(0, group_name)

        # We'll add a subgroup for the computation grid layers (to distinguish them from result layers)
        grid_group = layer_group.findGroup(GRID_GROUP_NAME)
        if not grid_group:
            grid_group = layer_group.insertGroup(0, GRID_GROUP_NAME)

        return layer_group

    @staticmethod
    def _layer_node_renamed(node: QgsLayerTreeNode, text: str, item: ThreeDiGridItem):
        if node is item.layer_group:
            item.setText(text)

    @staticmethod
    def _resolve_result_item_text(file: Path) -> str:
        """The text of the result item depends on its containing file structure
        """
        if file.parent is not None:
            return file.parent.stem

        # Fallback
        return file.stem

    @staticmethod
    def _resolve_grid_item_text(file: Path) -> str:
        """The text of the grid item depends on its containing file structure

        In case the grid file is in the 3Di Models & Simulations local directory
        structure, the text should be schematisation name + revision nr. Otherwise just a number.
        """
        if file.parent.parent is not None and file.parent.parent.parent is not None:
            folder = file.parent
            if folder.stem == "grid":
                rev_folder = folder.parent
                return rev_folder.parent.stem + " " + ThreeDiPluginLayerManager._retrieve_revision_str(rev_folder)

            folder = file.parent.parent
            if folder.stem == "results":
                rev_folder = folder.parent
                return rev_folder.parent.stem + " " + ThreeDiPluginLayerManager._retrieve_revision_str(rev_folder)

        # Fallback
        return file.parent.stem

    @staticmethod
    def _retrieve_revision_str(path: Path) -> str:
        """Retrieves the revision number from the path."""
        rev_folder = str(path.stem)
        if rev_folder.endswith("work in progress") :
            return "(WIP)"

        version = re.match("^revision (\d+)$", rev_folder)
        if version is not None:
            return "#" + version.group(1)

        # Fallback
        return "(None)"

1	import os	1✔
2	from collections import OrderedDict	1✔
3	from pathlib import Path	1✔
4	import re	1✔
5	import uuid	1✔
6	from qgis.PyQt.QtCore import Qt	1✔
7	from threedigrid.admin.exporters.geopackage import GeopackageExporter	1✔
8	from qgis.PyQt.QtCore import QObject, pyqtSlot, pyqtSignal, QVariant	1✔
9	from qgis.core import QgsVectorLayer, QgsProject, QgsMapLayer, QgsField, QgsWkbTypes, QgsLayerTreeNode	1✔
10	from threedi_results_analysis.threedi_plugin_model import ThreeDiGridItem, ThreeDiResultItem	1✔
11	from threedi_results_analysis.utils.constants import TOOLBOX_QGIS_GROUP_NAME, TOOLBOX_MESSAGE_TITLE	1✔
12	from threedi_results_analysis.utils.user_messages import StatusProgressBar, messagebar_message, pop_up_critical	1✔
13	from threedi_results_analysis.utils.utils import safe_join	1✔
14	from qgis.utils import iface	1✔
15
16	styles_dir = os.path.join(os.path.dirname(os.path.realpath(__file__)), "layer_styles", "grid")	1✔
17
18	import logging	1✔
19	logger = logging.getLogger(__name__)	1✔
20
21	GRID_GROUP_NAME = "Computational grid"	1✔
22
23
24	def dirty(func):	1✔
25	"""
26	This decorator ensures the QGIS project is marked as dirty when
27	the function is done.
28	"""
29	def wrapper(args, *kwargs):	1✔
30	func(args, *kwargs)	×
31	QgsProject.instance().setDirty()	×
32	return wrapper	1✔
33
34
35	def copy_layer_into_memory_layer(source_layer, layer_name, dest_layer):	1✔
36
37	source_provider = source_layer.dataProvider()	×
38
39	uri = "{0}?crs=EPSG:{1}".format(	×
40	QgsWkbTypes.displayString(source_provider.wkbType()).lstrip("WKB"),
41	str(source_provider.crs().postgisSrid()),
42	)
43
44	if dest_layer is None:	×
45	dest_layer = QgsVectorLayer(uri, layer_name, "memory")	×
46	else:
47	logger.info("Reusing memory layer instance")	×
48
49	dest_provider = dest_layer.dataProvider()	×
50
51	if not dest_provider.addAttributes(source_provider.fields()):	×
52	logger.error(dest_provider.lastError())	×
53	dest_layer.updateFields()	×
54
55	if not dest_provider.addFeatures(source_provider.getFeatures()):	×
56	logger.error(dest_provider.lastError())	×
57	dest_layer.updateExtents()	×
58
59	if source_provider.featureCount() != dest_provider.featureCount():	×
60	messagebar_message("Error", "Not all features are loaded in layer {layer_name}!", level=2, duration=5)	×
61
62	return dest_layer	×
63
64
65	# Layers need to be in specific order and naming:
66	gpkg_layers = OrderedDict(	1✔
67	[
68	("Model properties", "meta"),
69	("Pump (point)", "pump"),
70	("Node", "node"),
71	("Pump (line)", "pump_linestring"),
72	("Flowline", "flowline"),
73	("Cell", "cell"),
74	("Obstacle", "obstacle"),
75	]
76	)
77
78
79	class ThreeDiPluginLayerManager(QObject):	1✔
80	"""
81	The Layer manager creates layers from a geopackage and keeps track
82	of the connection between model items (grids) and layers.
83
84	In case a model item is deleted, the corresponding layers are also
85	deleted.
86	"""
87	grid_loaded = pyqtSignal(ThreeDiGridItem)	1✔
88	result_loaded = pyqtSignal(ThreeDiResultItem, ThreeDiGridItem)	1✔
89	grid_unloaded = pyqtSignal(ThreeDiGridItem)	1✔
90	result_unloaded = pyqtSignal(ThreeDiResultItem)	1✔
91
92	# error signals so the UI can deal with this accordingly
93	grid_not_loaded = pyqtSignal(ThreeDiGridItem)	1✔
94	result_not_loaded = pyqtSignal(ThreeDiResultItem, ThreeDiGridItem)	1✔
95
96	def __init__(self, args, *kwargs):	1✔
97	super().__init__(args, *kwargs)	×
98
99	@pyqtSlot(ThreeDiGridItem)	1✔
100	def load_grid(self, grid_item: ThreeDiGridItem) -> bool:	1✔
101	# generate geopackage if needed and point item path to it
102	if grid_item.path.suffix == ".h5":	×
103	path_h5 = grid_item.path	×
104	path_gpkg = path_h5.with_suffix(".gpkg")	×
105	if not path_gpkg.exists():	×
106	self.__class__._generate_gpkg(path_h5=path_h5, path_gpkg=path_gpkg)	×
107	grid_item.path = path_gpkg	×
108	else:
109	path_gpkg = grid_item.path	×
110
111	# Text of item is determined by folder structure
112	if not grid_item.text():	×
113	grid_item.setText(ThreeDiPluginLayerManager._resolve_grid_item_text(grid_item.path))	×
114
115	if not ThreeDiPluginLayerManager._add_layers_from_gpkg(path_gpkg, grid_item):	×
116	pop_up_critical("Failed adding the layers to the project.")	×
117	self.grid_not_loaded.emit(grid_item)	×
118	return False	×
119
120	messagebar_message(TOOLBOX_MESSAGE_TITLE, "Added layers to the project", duration=2)	×
121
122	self.grid_loaded.emit(grid_item)	×
123	return True	×
124
125	@pyqtSlot(ThreeDiGridItem)	1✔
126	def unload_grid(self, item: ThreeDiGridItem) -> bool:	1✔
127	"""Removes the corresponding layers from the group in the project"""
128
129	# It could be possible that some layers have been dragged outside the
130	# layer group. Delete the individual layers first
131	for layer_id in item.layer_ids.values():	×
132	assert QgsProject.instance().mapLayer(layer_id)	×
133	QgsProject.instance().removeMapLayer(layer_id)	×
134	item.layer_ids.clear()	×
135
136	# Deletion of root node of a tree will delete all nodes of the tree.
137	# In case the user dragged another layer in the group, remove the reference
138	# from this grid to the group, but don't delete it from QGIS.
139	assert item.layer_group	×
140	grid_group = item.layer_group.findGroup(GRID_GROUP_NAME)	×
141
142	# Remove "Computational Grid" group
143	if len(grid_group.children()) == 0:	×
144	item.layer_group.removeChildNode(grid_group)	×
145
146	if len(item.layer_group.children()) == 0:	×
147	item.layer_group.parent().removeChildNode(item.layer_group)	×
148	else:
149	logger.info(f"Item group of grid {item.text()} contains external layers: not removing.")	×
150
151	item.layer_group = None	×
152	iface.mapCanvas().refresh()	×
153
154	self.grid_unloaded.emit(item)	×
155
156	@dirty	1✔
157	@pyqtSlot(ThreeDiGridItem)	1✔
158	def update_grid(self, item: ThreeDiGridItem) -> bool:	1✔
159	"""Updates the group name in the project"""
160	assert item.layer_group	×
161	item.layer_group.setName(item.text())	×
162	return True	×
163
164	@pyqtSlot(ThreeDiResultItem, ThreeDiGridItem)	1✔
165	def load_result(self, threedi_result_item: ThreeDiResultItem, grid_item: ThreeDiGridItem) -> bool:	1✔
166
167	# Text of item is determined by folder structure
168	if not threedi_result_item.text():	×
169	threedi_result_item.setText(ThreeDiPluginLayerManager._resolve_result_item_text(threedi_result_item.path))	×
170
171	# Add result fields for this result to the grid layers
172	logger.info("Adding result fields to grid layers")	×
173	for layer_id in grid_item.layer_ids.values():	×
174	layer = QgsProject.instance().mapLayer(layer_id)	×
175	provider = layer.dataProvider()	×
176
177	# Generate a random field name, with the result text
178	# as alias (display) name.
179	unique_identifier = str(uuid.uuid4())	×
180
181	result_field_name = "result_" + unique_identifier	×
182	result_field = QgsField(result_field_name, QVariant.Double)	×
183	result_field.setAlias(threedi_result_item.text())	×
184
185	initial_value_field_name = "initial_value_" + unique_identifier	×
186	initial_value_field = QgsField(initial_value_field_name, QVariant.Double)	×
187	initial_value_field.setAlias(threedi_result_item.text() + "_initial_value")	×
188
189	# Check for duplicate field names (even though QGIS does not allow
190	# addition of QgsFields (both attribute or expression) with already
191	# existing names AND the generated layers are marked READONLY)
192	if (layer.fields().indexFromName(result_field_name) != -1 or	×
193	layer.fields().indexFromName(initial_value_field_name) != -1):
194	logger.error("Field already exist, aborting addition.")	×
195	self.result_not_loaded.emit(threedi_result_item, grid_item)	×
196	return False	×
197
198	provider.addAttributes([result_field, initial_value_field])	×
199	layer.updateFields()	×
200
201	# Store the added field names so we can remove the field when the result is removed
202	threedi_result_item._result_field_names[layer_id] = (result_field_name, initial_value_field_name)	×
203
204	self.result_loaded.emit(threedi_result_item, grid_item)	×
205	return True	×
206
207	@pyqtSlot(ThreeDiResultItem)	1✔
208	def unload_result(self, threedi_result_item: ThreeDiResultItem) -> bool:	1✔
209	# Remove the corresponding result fields from the grid layers
210	for layer_id, result_field_names in threedi_result_item._result_field_names.items():	×
211	# It could be that the map layer is removed by QGIS
212	if QgsProject.instance().mapLayer(layer_id) is not None:	×
213	layer = QgsProject.instance().mapLayer(layer_id)	×
214	provider = layer.dataProvider()	×
215
216	assert len(result_field_names) == 2	×
217	idx = layer.fields().indexFromName(result_field_names[0])	×
218	assert idx != -1	×
219	provider.deleteAttributes([idx])	×
220	layer.updateFields()	×
221
222	idx = layer.fields().indexFromName(result_field_names[1])	×
223	assert idx != -1	×
224	provider.deleteAttributes([idx])	×
225	layer.updateFields()	×
226
227	threedi_result_item._result_field_names.clear()	×
228
229	# In case the corresponding grid contains no more (selected) results, the
230	# original styling should be reset
231	reset_styling = True	×
232	grid_item = threedi_result_item.parent()	×
233	assert isinstance(grid_item, ThreeDiGridItem)	×
234	if grid_item.hasChildren():	×
235	for i in range(grid_item.rowCount()):	×
236	result_item = grid_item.child(i)	×
NEW 237	if (result_item.checkState() == Qt.CheckState.Checked and threedi_result_item is not result_item):	×
238	reset_styling = False	×
239
240	if reset_styling:	×
241	self.reset_styling(grid_item)	×
242
243	self.result_unloaded.emit(threedi_result_item)	×
244	return True	×
245
246	@dirty	1✔
247	@pyqtSlot(ThreeDiResultItem)	1✔
248	def result_unchecked(self, item: ThreeDiResultItem):	1✔
249	# In case all results are unchecked, revert back to default styling (and naming)
250	grid_item = item.parent()	×
251	assert isinstance(grid_item, ThreeDiGridItem)	×
252	if grid_item.hasChildren():	×
253	for i in range(grid_item.rowCount()):	×
254	result_item = grid_item.child(i)	×
NEW 255	if result_item.checkState() == Qt.CheckState.Checked:	×
256	return	×
257
258	self.reset_styling(grid_item)	×
259
260	@pyqtSlot(ThreeDiGridItem)	1✔
261	def reset_styling(self, grid_item: ThreeDiGridItem) -> None:	1✔
262	"""Sets all the grid layers for a given grid back to their original name and style"""
263	for layer_name, table_name in gpkg_layers.items():	×
264
265	# Some models do not contain pump or obstacle layers.
266	if table_name not in grid_item.layer_ids.keys():	×
267	continue	×
268
269	scratch_layer = QgsProject.instance().mapLayer(grid_item.layer_ids[table_name])	×
270	assert scratch_layer	×
271
272	# (Re)apply the style and naming
273	qml_path = safe_join(styles_dir, f"{table_name}.qml")	×
274	if os.path.exists(qml_path):	×
275	msg, res = scratch_layer.loadNamedStyle(qml_path)	×
276	if not res:	×
277	logger.error(f"Unable to load style: {msg}")	×
278
279	scratch_layer.setName(layer_name)	×
280	iface.layerTreeView().refreshLayerSymbology(scratch_layer.id())	×
281	scratch_layer.triggerRepaint()	×
282
283	@dirty	1✔
284	@pyqtSlot(ThreeDiResultItem)	1✔
285	def update_result(self, threedi_result_item: ThreeDiResultItem) -> bool:	1✔
286	# Update the display name of the result fields
287	logger.info("Updating result fields")	×
288	for layer_id, result_field_names in threedi_result_item._result_field_names.items():	×
289	layer = QgsProject.instance().mapLayer(layer_id)	×
290
291	assert len(result_field_names) == 2	×
292	idx = layer.fields().indexFromName(result_field_names[0])	×
293	assert idx != -1	×
294	layer.setFieldAlias(idx, threedi_result_item.text())	×
295
296	idx = layer.fields().indexFromName(result_field_names[1])	×
297	assert idx != -1	×
298	layer.setFieldAlias(idx, threedi_result_item.text() + '_initial_value')	×
299
300	return True	×
301
302	@staticmethod	1✔
303	def _generate_gpkg(path_h5, path_gpkg) -> None:	1✔
304	progress_bar = StatusProgressBar(100, "Generating computational grid geopackage")	×
305	exporter = GeopackageExporter(path_h5.open('rb'), str(path_gpkg))	×
306	exporter.export(	×
307	lambda count, total, pb=progress_bar: pb.set_value((count * 100) // total)
308	)
309	del progress_bar	×
310
311	messagebar_message(TOOLBOX_MESSAGE_TITLE, "Generated computational grid geopackage")	×
312
313	@staticmethod	1✔
314	def _add_layers_from_gpkg(path, item: ThreeDiGridItem) -> bool:	1✔
315	"""
316	Retrieves (a subset of the) layers from gpk and add to project.
317	"""
318
319	invalid_layers = []	×
320	empty_layers = []	×
321
322	item.layer_group = ThreeDiPluginLayerManager._get_or_create_group(item.text())	×
323
324	# Use to modify grid name when LayerGroup is renamed
325	item.layer_group.nameChanged.connect(lambda node, txt, grid_item=item: ThreeDiPluginLayerManager._layer_node_renamed(node, txt, grid_item))	×
326
327	progress_bar = StatusProgressBar(len(gpkg_layers) - 1, "Adding computational grid layers")	×
328	for layer_name, table_name in gpkg_layers.items():	×
329
330	# QGIS does save memory layers to the project file (but without the data)
331	# Removing the scratch layer and resaving the project causes QGIS to crash,
332	# therefore we reuse the layer instance.
333	scratch_layer = None	×
334	if table_name in item.layer_ids.keys():	×
335	scratch_layer = QgsProject.instance().mapLayer(item.layer_ids[table_name])	×
336	if scratch_layer:	×
337	logger.info(f"Map layer corresponding to table {item.layer_ids[table_name]} already exist in project, reusing...")	×
338
339	# Using the QgsInterface function addVectorLayer shows (annoying) confirmation dialogs
340	# iface.addVectorLayer(gpkg_file + "\|layername=" + layer, layer, 'ogr')
341	vector_layer = QgsVectorLayer(str(path) + "\|layername=" + table_name, layer_name, "ogr")	×
342	if not vector_layer.isValid():	×
343	invalid_layers.append(layer_name)	×
344	continue	×
345
346	# Only load layers that contain some features
347	if not vector_layer.featureCount():	×
348	empty_layers.append(layer_name)	×
349	continue	×
350
351	vector_layer = copy_layer_into_memory_layer(	×
352	vector_layer, layer_name, scratch_layer
353	)
354
355	# Apply the style
356	qml_path = safe_join(styles_dir, f"{table_name}.qml")	×
357	if os.path.exists(qml_path):	×
358	msg, res = vector_layer.loadNamedStyle(qml_path)	×
359	if not res:	×
360	logger.error(f"Unable to load style: {msg}")	×
361	# prior to QGIS 3.24, saveStyleToDatabase would show an (annoying) message box
362	# warning when a style with the same styleName already existed. Unfortunately,
363	# QgsProviderRegistry::styleExists is not available in Python
364	# if table_name not in vector_layer.listStylesInDatabase()[2]:
365	# Memory providers do not support saving of styles, commented
366	# msg = vector_layer.saveStyleToDatabase(table_name, "", True, "")
367	# if msg:
368	# logger.error(f"Unable to save style to DB: {msg}")
369
370	if not vector_layer.setReadOnly(True):	×
371	logger.error(f"Unable to set layer {table_name} to read-only")	×
NEW 372	vector_layer.setFlags(QgsMapLayer.LayerFlag.Searchable \| QgsMapLayer.LayerFlag.Identifiable)	×
373	# Require to keep track of the purpose of the layer (node, pump, flowline etc)
374	# Can also be used to check whether this layer is generated by the Toolbox
375	vector_layer.setObjectName(table_name)	×
376
377	if scratch_layer is None:	×
378	# Keep track of layer id for future reference (deletion of grid item)
379	item.layer_ids[table_name] = vector_layer.id()	×
380
381	QgsProject.instance().addMapLayer(vector_layer, addToLegend=False)	×
382	# Add to computational grid subgroup (created above)
383	item.layer_group.findGroup(GRID_GROUP_NAME).addLayer(vector_layer)	×
384
385	progress_bar.increase_progress()	×
386
387	del progress_bar	×
388
389	# Invalid layers info
390	if invalid_layers:	×
391	logger.warning("The following layers are missing or invalid:\n * " + "\n * ".join(invalid_layers) + "\n\n")	×
392
393	# Empty layers info
394	if empty_layers:	×
395	logger.warning("The following layers contained no feature:\n * " + "\n * ".join(empty_layers) + "\n\n")	×
396
397	return True	×
398
399	@staticmethod	1✔
400	def _get_or_create_group(group_name: str):	1✔
401	root = QgsProject.instance().layerTreeRoot()	×
402	root_group = root.findGroup(TOOLBOX_QGIS_GROUP_NAME)	×
403	if not root_group:	×
404	root_group = root.insertGroup(0, TOOLBOX_QGIS_GROUP_NAME)	×
405
406	layer_group = root_group.findGroup(group_name)	×
407	if not layer_group:	×
408	layer_group = root_group.insertGroup(0, group_name)	×
409
410	# We'll add a subgroup for the computation grid layers (to distinguish them from result layers)
411	grid_group = layer_group.findGroup(GRID_GROUP_NAME)	×
412	if not grid_group:	×
413	grid_group = layer_group.insertGroup(0, GRID_GROUP_NAME)	×
414
415	return layer_group	×
416
417	@staticmethod	1✔
418	def _layer_node_renamed(node: QgsLayerTreeNode, text: str, item: ThreeDiGridItem):	1✔
419	if node is item.layer_group:	×
420	item.setText(text)	×
421
422	@staticmethod	1✔
423	def _resolve_result_item_text(file: Path) -> str:	1✔
424	"""The text of the result item depends on its containing file structure
425	"""
426	if file.parent is not None:	×
427	return file.parent.stem	×
428
429	# Fallback
430	return file.stem	×
431
432	@staticmethod	1✔
433	def _resolve_grid_item_text(file: Path) -> str:	1✔
434	"""The text of the grid item depends on its containing file structure
435
436	In case the grid file is in the 3Di Models & Simulations local directory
437	structure, the text should be schematisation name + revision nr. Otherwise just a number.
438	"""
439	if file.parent.parent is not None and file.parent.parent.parent is not None:	×
440	folder = file.parent	×
441	if folder.stem == "grid":	×
442	rev_folder = folder.parent	×
443	return rev_folder.parent.stem + " " + ThreeDiPluginLayerManager._retrieve_revision_str(rev_folder)	×
444
445	folder = file.parent.parent	×
446	if folder.stem == "results":	×
447	rev_folder = folder.parent	×
448	return rev_folder.parent.stem + " " + ThreeDiPluginLayerManager._retrieve_revision_str(rev_folder)	×
449
450	# Fallback
451	return file.parent.stem	×
452
453	@staticmethod	1✔
454	def _retrieve_revision_str(path: Path) -> str:	1✔
455	"""Retrieves the revision number from the path."""
456	rev_folder = str(path.stem)	×
457	if rev_folder.endswith("work in progress") :	×
458	return "(WIP)"	×
459
460	version = re.match("^revision (\d+)$", rev_folder)	×
461	if version is not None:	×
462	return "#" + version.group(1)	×
463
464	# Fallback
465	return "(None)"	×

nens / ThreeDiToolbox / #2589

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