#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

81.62

/threedi_plugin_model_validation.py

from qgis.PyQt.QtCore import QObject, pyqtSignal, pyqtSlot
from pathlib import Path
from qgis.core import Qgis, QgsVectorLayer
from threedi_results_analysis.threedi_plugin_model import ThreeDiGridItem, ThreeDiResultItem
from threedi_results_analysis.utils.user_messages import messagebar_message, pop_up_critical
from threedi_results_analysis.threedi_plugin_model import ThreeDiPluginModel
from threedi_results_analysis.utils.constants import TOOLBOX_MESSAGE_TITLE
from threedi_results_analysis.utils.utils import listdirs
import h5py
import logging
logger = logging.getLogger(__name__)


class ThreeDiPluginModelValidator(QObject):
    """
    This class validates 3Di computation grid and result files. When
    a grid or result is valid, a signal is emited
    so listeners can handle accordingly.
    """
    grid_valid = pyqtSignal(ThreeDiGridItem)
    result_valid = pyqtSignal(ThreeDiResultItem, ThreeDiGridItem)
    grid_invalid = pyqtSignal(ThreeDiGridItem)
    result_invalid = pyqtSignal(ThreeDiResultItem, ThreeDiGridItem)

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

    @pyqtSlot(str)
    def validate_grid(self, grid_file: str, result_slug: str = None) -> ThreeDiGridItem:
        """
        Validates the grid and returns the new (or already existing) ThreeDiGridItem. Also emits signal.

        if required_slug is not None, the first grid in the model with this slug will be selected.

        If not, the validator will check whether a grid with the same slug is found as the grid_file.

        If not, the validor will create a new ThreeDiGridItem and emit the grid_valid signal.
        """
        logger.info(f"Validate_grid({grid_file}, {result_slug}")

        # First check whether a grid with the result_slug already exists
        if result_slug:
            for grid in self.model.get_grids():
                # Check whether corresponding grid item belongs to same model as result
                other_grid_model_slug = ThreeDiPluginModelValidator.get_grid_slug(Path(grid.path))
                if result_slug == other_grid_model_slug:
                    print("using result slug")
                    messagebar_message(TOOLBOX_MESSAGE_TITLE, "Result attached to computational grid that was already loaded.", Qgis.MessageLevel.Info, 5)
                    return grid

        # Check whether the model already contains a grid with the new grid files slug
        if grid_file:
            grid_model_slug = ThreeDiPluginModelValidator.get_grid_slug(Path(grid_file))
            if grid_model_slug:
                for grid in self.model.get_grids():
                    # Check whether corresponding grid item belongs to same model as result
                    other_grid_model_slug = ThreeDiPluginModelValidator.get_grid_slug(Path(grid.path))
                    if grid_model_slug == other_grid_model_slug:
                        print("using grid slug")
                        messagebar_message(TOOLBOX_MESSAGE_TITLE, "Result attached to computational grid that was already loaded.", Qgis.MessageLevel.Info, 5)
                        return grid

        if not grid_file:
            messagebar_message(TOOLBOX_MESSAGE_TITLE, "No computational grid for this result could be found, aborting", Qgis.MessageLevel.Critical, 5)
            return None

        # Check whether model already contains this grid file.
        for i in range(self.model.invisibleRootItem().rowCount()):
            grid_item = self.model.invisibleRootItem().child(i)
            if grid_item.path.with_suffix("") == Path(grid_file).with_suffix(""):
                messagebar_message(TOOLBOX_MESSAGE_TITLE, "This computional grid was already loaded.", Qgis.MessageLevel.Info, 5)
                self.grid_invalid.emit(ThreeDiGridItem(Path(grid_file), ""))
                return grid_item

        # Note that in the 3Di M&S working directory setup, each results
        # folder in the revision can contain the same gridadmin file. Check
        # whether there is a grid loaded from one of these result folders.
        folder = Path(grid_file).parent
        if folder.parent.name == 'results':
            if str(folder.parent.parent.name).startswith('revision'):
                result_folders = [Path(d) for d in listdirs(folder.parent)]

                # Iterate over the grids
                for i in range(self.model.invisibleRootItem().rowCount()):
                    grid_item = self.model.invisibleRootItem().child(i)
                    assert isinstance(grid_item, ThreeDiGridItem)
                    grid_folder = Path(grid_item.path).parent
                    if grid_folder in result_folders:
                        messagebar_message(TOOLBOX_MESSAGE_TITLE, "This computional grid was already loaded.", Qgis.MessageLevel.Info, 5)
                        # Todo: should we do a simple shallow file-compare?
                        self.grid_invalid.emit(grid_item)
                        return grid_item

        new_grid = ThreeDiGridItem(Path(grid_file), "")
        self.grid_valid.emit(new_grid)
        return new_grid

    @pyqtSlot(str, str)
    def validate_result_grid(self, results_path: str, grid_path: str):
        """
        Validate the result, but first validate (and add) the grid.
        """
        # First check whether this is the right grid, or a more appropriate is already
        # in the model (with same slug)
        result_model_slug = ThreeDiPluginModelValidator.get_result_slug(Path(results_path))
        logger.info(f"Validating {results_path} ({result_model_slug}) and {grid_path}")
        grid_item = self.validate_grid(grid_path, result_model_slug)
        if not grid_item:
            messagebar_message(TOOLBOX_MESSAGE_TITLE, "No computational grid for this result could be found, aborting", Qgis.MessageLevel.Critical, 5)
            return

        self._validate_result(results_path, grid_item)

    def _validate_result(self, results_path: str, grid_item: ThreeDiGridItem) -> bool:
        logger.info(f"Validating result with grid item {grid_item.text()}")
        """
        Validate the result when added to the selected grid item. Returns True
        on success and emits result_valid or result_invalid.
        """
        def fail(msg):
            messagebar_message(TOOLBOX_MESSAGE_TITLE, msg, Qgis.MessageLevel.Warning, 5)
            self.result_invalid.emit(result_item, grid_item)
            return False

        result_item = ThreeDiResultItem(Path(results_path))

        if self.model.contains(Path(results_path), True):
            return fail("This result was already loaded")

        # Check correct file name
        if not result_item.path.name == "results_3di.nc":
            return fail("Unexpected file name for results file")

        # Check opening with h5py, detects a.o. incomplete downloads
        try:
            results_h5 = h5py.File(result_item.path.open("rb"), "r")
        except OSError as error:
            if "truncated file" in str(error):
                return fail(
                    f"Results file {result_item.path} is incomplete. "
                    "If possible, copy or download it again."
                )
            return fail(f"Results file cannot be opened: {str(error.errno)} {str(error.strerror)} {str(error.args)}")

        # Try to open accompanying aggregate results file
        aggregate_results_path = result_item.path.with_name("aggregate_results_3di.nc")
        if aggregate_results_path.exists():
            try:
                h5py.File(aggregate_results_path.open("rb"), "r")
            except OSError as error:
                if "truncated file" in str(error):
                    return fail(
                        f"Aggregate results file {aggregate_results_path} "
                        "is incomplete. If possible, copy or download it "
                        "again."
                    )
                return fail("Aggregate results file cannot be opened.")

        # Any modern enough calc core adds a 'threedicore_version' atribute
        if "threedicore_version" not in results_h5.attrs:
            return fail("Result file is too old and cannot be opened with 3Di Result Analysis.")

        # Check whether corresponding grid item belongs to same model as result
        result_model_slug = ThreeDiPluginModelValidator.get_result_slug(result_item.path)

        grid_model_slug = ThreeDiPluginModelValidator.get_grid_slug(grid_item.path)
        print(f"Comparing grid slug: {grid_model_slug} to result slug: {result_model_slug}")
        logger.info(f"Comparing grid slug: {grid_model_slug} to result slug: {result_model_slug}")

        if not grid_model_slug or not result_model_slug:
            msg = "No grid or result slug available, unable to validate to which 3Di model this computational grid or result belongs"
            messagebar_message(TOOLBOX_MESSAGE_TITLE, msg, Qgis.MessageLevel.Warning, 5)
        elif result_model_slug != grid_model_slug:
            # Really wrong grid, find a grid with the right slug, if not available, abort with pop-up
            root_node = grid_item.model().invisibleRootItem()
            for i in range(root_node.rowCount()):
                other_grid_item = root_node.child(i)
                other_grid_model_slug = ThreeDiPluginModelValidator.get_grid_slug(other_grid_item.path)

                print("comparing result slug to other grids again")
                if result_model_slug == other_grid_model_slug:
                    messagebar_message(TOOLBOX_MESSAGE_TITLE, "Result attached to computational grid that was already loaded.", Qgis.MessageLevel.Warning, 5)

                    # Propagate the result with the new parent grid
                    self.result_valid.emit(result_item, other_grid_item)
                    return True

            msg = "No computational grid for this result could be found, cannot load this result."
            pop_up_critical(msg)
            return fail(msg)

        self.result_valid.emit(result_item, grid_item)
        return True

    @staticmethod
    def get_grid_slug(geopackage_path: Path) -> str:
        meta_layer = QgsVectorLayer(str(geopackage_path.with_suffix(".gpkg")) + "|layername=meta", "meta", "ogr")

        if not meta_layer.isValid() or not (meta_layer.featureCount() == 1):
            logger.warning("Invalid, zero or more than 1 meta data table. Unable to derive slug")
            return None

        # Take first
        meta = next(meta_layer.getFeatures())
        return meta["model_slug"]

    @staticmethod
    def get_result_slug(netcdf_path: Path) -> str:
        results_h5 = h5py.File(netcdf_path.open("rb"), "r")
        result_model_slug = results_h5.attrs['model_slug'].decode()
        if result_model_slug == "NO SLUG FOUND":
            result_model_slug = ""

        return result_model_slug

1	from qgis.PyQt.QtCore import QObject, pyqtSignal, pyqtSlot	1✔
2	from pathlib import Path	1✔
3	from qgis.core import Qgis, QgsVectorLayer	1✔
4	from threedi_results_analysis.threedi_plugin_model import ThreeDiGridItem, ThreeDiResultItem	1✔
5	from threedi_results_analysis.utils.user_messages import messagebar_message, pop_up_critical	1✔
6	from threedi_results_analysis.threedi_plugin_model import ThreeDiPluginModel	1✔
7	from threedi_results_analysis.utils.constants import TOOLBOX_MESSAGE_TITLE	1✔
8	from threedi_results_analysis.utils.utils import listdirs	1✔
9	import h5py	1✔
10	import logging	1✔
11	logger = logging.getLogger(__name__)	1✔
12
13
14	class ThreeDiPluginModelValidator(QObject):	1✔
15	"""
16	This class validates 3Di computation grid and result files. When
17	a grid or result is valid, a signal is emited
18	so listeners can handle accordingly.
19	"""
20	grid_valid = pyqtSignal(ThreeDiGridItem)	1✔
21	result_valid = pyqtSignal(ThreeDiResultItem, ThreeDiGridItem)	1✔
22	grid_invalid = pyqtSignal(ThreeDiGridItem)	1✔
23	result_invalid = pyqtSignal(ThreeDiResultItem, ThreeDiGridItem)	1✔
24
25	def __init__(self, model: ThreeDiPluginModel, args, *kwargs):	1✔
26	self.model = model	1✔
27	super().__init__(args, *kwargs)	1✔
28
29	@pyqtSlot(str)	1✔
30	def validate_grid(self, grid_file: str, result_slug: str = None) -> ThreeDiGridItem:	1✔
31	"""
32	Validates the grid and returns the new (or already existing) ThreeDiGridItem. Also emits signal.
33
34	if required_slug is not None, the first grid in the model with this slug will be selected.
35
36	If not, the validator will check whether a grid with the same slug is found as the grid_file.
37
38	If not, the validor will create a new ThreeDiGridItem and emit the grid_valid signal.
39	"""
40	logger.info(f"Validate_grid({grid_file}, {result_slug}")	1✔
41
42	# First check whether a grid with the result_slug already exists
43	if result_slug:	1✔
44	for grid in self.model.get_grids():	1✔
45	# Check whether corresponding grid item belongs to same model as result
46	other_grid_model_slug = ThreeDiPluginModelValidator.get_grid_slug(Path(grid.path))	1✔
47	if result_slug == other_grid_model_slug:	1✔
48	print("using result slug")	1✔
49	messagebar_message(TOOLBOX_MESSAGE_TITLE, "Result attached to computational grid that was already loaded.", Qgis.MessageLevel.Info, 5)	1✔
50	return grid	1✔
51
52	# Check whether the model already contains a grid with the new grid files slug
53	if grid_file:	1✔
54	grid_model_slug = ThreeDiPluginModelValidator.get_grid_slug(Path(grid_file))	1✔
55	if grid_model_slug:	1✔
56	for grid in self.model.get_grids():	1✔
57	# Check whether corresponding grid item belongs to same model as result
58	other_grid_model_slug = ThreeDiPluginModelValidator.get_grid_slug(Path(grid.path))	1✔
59	if grid_model_slug == other_grid_model_slug:	1✔
60	print("using grid slug")	1✔
61	messagebar_message(TOOLBOX_MESSAGE_TITLE, "Result attached to computational grid that was already loaded.", Qgis.MessageLevel.Info, 5)	1✔
62	return grid	1✔
63
64	if not grid_file:	1✔
65	messagebar_message(TOOLBOX_MESSAGE_TITLE, "No computational grid for this result could be found, aborting", Qgis.MessageLevel.Critical, 5)	1✔
66	return None	1✔
67
68	# Check whether model already contains this grid file.
69	for i in range(self.model.invisibleRootItem().rowCount()):	1✔
70	grid_item = self.model.invisibleRootItem().child(i)	1✔
71	if grid_item.path.with_suffix("") == Path(grid_file).with_suffix(""):	1✔
72	messagebar_message(TOOLBOX_MESSAGE_TITLE, "This computional grid was already loaded.", Qgis.MessageLevel.Info, 5)	1✔
73	self.grid_invalid.emit(ThreeDiGridItem(Path(grid_file), ""))	1✔
74	return grid_item	1✔
75
76	# Note that in the 3Di M&S working directory setup, each results
77	# folder in the revision can contain the same gridadmin file. Check
78	# whether there is a grid loaded from one of these result folders.
79	folder = Path(grid_file).parent	1✔
80	if folder.parent.name == 'results':	1✔
81	if str(folder.parent.parent.name).startswith('revision'):	×
82	result_folders = [Path(d) for d in listdirs(folder.parent)]	×
83
84	# Iterate over the grids
85	for i in range(self.model.invisibleRootItem().rowCount()):	×
86	grid_item = self.model.invisibleRootItem().child(i)	×
87	assert isinstance(grid_item, ThreeDiGridItem)	×
88	grid_folder = Path(grid_item.path).parent	×
89	if grid_folder in result_folders:	×
NEW 90	messagebar_message(TOOLBOX_MESSAGE_TITLE, "This computional grid was already loaded.", Qgis.MessageLevel.Info, 5)	×
91	# Todo: should we do a simple shallow file-compare?
92	self.grid_invalid.emit(grid_item)	×
93	return grid_item	×
94
95	new_grid = ThreeDiGridItem(Path(grid_file), "")	1✔
96	self.grid_valid.emit(new_grid)	1✔
97	return new_grid	1✔
98
99	@pyqtSlot(str, str)	1✔
100	def validate_result_grid(self, results_path: str, grid_path: str):	1✔
101	"""
102	Validate the result, but first validate (and add) the grid.
103	"""
104	# First check whether this is the right grid, or a more appropriate is already
105	# in the model (with same slug)
106	result_model_slug = ThreeDiPluginModelValidator.get_result_slug(Path(results_path))	1✔
107	logger.info(f"Validating {results_path} ({result_model_slug}) and {grid_path}")	1✔
108	grid_item = self.validate_grid(grid_path, result_model_slug)	1✔
109	if not grid_item:	1✔
NEW 110	messagebar_message(TOOLBOX_MESSAGE_TITLE, "No computational grid for this result could be found, aborting", Qgis.MessageLevel.Critical, 5)	×
111	return	×
112
113	self._validate_result(results_path, grid_item)	1✔
114
115	def _validate_result(self, results_path: str, grid_item: ThreeDiGridItem) -> bool:	1✔
116	logger.info(f"Validating result with grid item {grid_item.text()}")	1✔
117	"""	1✔
118	Validate the result when added to the selected grid item. Returns True
119	on success and emits result_valid or result_invalid.
120	"""
121	def fail(msg):	1✔
122	messagebar_message(TOOLBOX_MESSAGE_TITLE, msg, Qgis.MessageLevel.Warning, 5)	1✔
123	self.result_invalid.emit(result_item, grid_item)	1✔
124	return False	1✔
125
126	result_item = ThreeDiResultItem(Path(results_path))	1✔
127
128	if self.model.contains(Path(results_path), True):	1✔
129	return fail("This result was already loaded")	1✔
130
131	# Check correct file name
132	if not result_item.path.name == "results_3di.nc":	1✔
133	return fail("Unexpected file name for results file")	1✔
134
135	# Check opening with h5py, detects a.o. incomplete downloads
136	try:	1✔
137	results_h5 = h5py.File(result_item.path.open("rb"), "r")	1✔
138	except OSError as error:	×
139	if "truncated file" in str(error):	×
140	return fail(	×
141	f"Results file {result_item.path} is incomplete. "
142	"If possible, copy or download it again."
143	)
144	return fail(f"Results file cannot be opened: {str(error.errno)} {str(error.strerror)} {str(error.args)}")	×
145
146	# Try to open accompanying aggregate results file
147	aggregate_results_path = result_item.path.with_name("aggregate_results_3di.nc")	1✔
148	if aggregate_results_path.exists():	1✔
149	try:	1✔
150	h5py.File(aggregate_results_path.open("rb"), "r")	1✔
151	except OSError as error:	×
152	if "truncated file" in str(error):	×
153	return fail(	×
154	f"Aggregate results file {aggregate_results_path} "
155	"is incomplete. If possible, copy or download it "
156	"again."
157	)
158	return fail("Aggregate results file cannot be opened.")	×
159
160	# Any modern enough calc core adds a 'threedicore_version' atribute
161	if "threedicore_version" not in results_h5.attrs:	1✔
162	return fail("Result file is too old and cannot be opened with 3Di Result Analysis.")	1✔
163
164	# Check whether corresponding grid item belongs to same model as result
165	result_model_slug = ThreeDiPluginModelValidator.get_result_slug(result_item.path)	1✔
166
167	grid_model_slug = ThreeDiPluginModelValidator.get_grid_slug(grid_item.path)	1✔
168	print(f"Comparing grid slug: {grid_model_slug} to result slug: {result_model_slug}")	1✔
169	logger.info(f"Comparing grid slug: {grid_model_slug} to result slug: {result_model_slug}")	1✔
170
171	if not grid_model_slug or not result_model_slug:	1✔
172	msg = "No grid or result slug available, unable to validate to which 3Di model this computational grid or result belongs"	×
NEW 173	messagebar_message(TOOLBOX_MESSAGE_TITLE, msg, Qgis.MessageLevel.Warning, 5)	×
174	elif result_model_slug != grid_model_slug:	1✔
175	# Really wrong grid, find a grid with the right slug, if not available, abort with pop-up
176	root_node = grid_item.model().invisibleRootItem()	1✔
177	for i in range(root_node.rowCount()):	1✔
178	other_grid_item = root_node.child(i)	1✔
179	other_grid_model_slug = ThreeDiPluginModelValidator.get_grid_slug(other_grid_item.path)	1✔
180
181	print("comparing result slug to other grids again")	1✔
182	if result_model_slug == other_grid_model_slug:	1✔
183	messagebar_message(TOOLBOX_MESSAGE_TITLE, "Result attached to computational grid that was already loaded.", Qgis.MessageLevel.Warning, 5)	1✔
184
185	# Propagate the result with the new parent grid
186	self.result_valid.emit(result_item, other_grid_item)	1✔
187	return True	1✔
188
189	msg = "No computational grid for this result could be found, cannot load this result."	1✔
190	pop_up_critical(msg)	1✔
191	return fail(msg)	1✔
192
193	self.result_valid.emit(result_item, grid_item)	1✔
194	return True	1✔
195
196	@staticmethod	1✔
197	def get_grid_slug(geopackage_path: Path) -> str:	1✔
198	meta_layer = QgsVectorLayer(str(geopackage_path.with_suffix(".gpkg")) + "\|layername=meta", "meta", "ogr")	1✔
199
200	if not meta_layer.isValid() or not (meta_layer.featureCount() == 1):	1✔
201	logger.warning("Invalid, zero or more than 1 meta data table. Unable to derive slug")	1✔
202	return None	1✔
203
204	# Take first
205	meta = next(meta_layer.getFeatures())	×
206	return meta["model_slug"]	×
207
208	@staticmethod	1✔
209	def get_result_slug(netcdf_path: Path) -> str:	1✔
210	results_h5 = h5py.File(netcdf_path.open("rb"), "r")	1✔
211	result_model_slug = results_h5.attrs['model_slug'].decode()	1✔
212	if result_model_slug == "NO SLUG FOUND":	1✔
213	result_model_slug = ""	×
214
215	return result_model_slug	1✔

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