diff --git a/CHANGES.rst b/CHANGES.rst index 74db201d..e7343a17 100755 --- a/CHANGES.rst +++ b/CHANGES.rst @@ -12,6 +12,7 @@ - Statistics: Add two water on street presets. - General: added some layer loading feedback. +- Passed handling of the 3Di working directory structure to `threedi_mi_utils` package. 3.1.12 (2023-10-19) diff --git a/dependencies.py b/dependencies.py index 3b5f04d8..cc6cf1ed 100644 --- a/dependencies.py +++ b/dependencies.py @@ -75,6 +75,7 @@ Dependency("hydxlib", "hydxlib", "==1.5.1", False), Dependency("h5netcdf", "h5netcdf", "", False), Dependency("greenlet", "greenlet", "!=0.4.17", False), + Dependency("threedi-mi-utils", "threedi_mi_utils", "==0.1.2", False), ] # On Windows, the hdf5 binary and thus h5py version depends on the QGis version diff --git a/external-dependencies/populate.sh b/external-dependencies/populate.sh index 931e2255..c1251c5e 100755 --- a/external-dependencies/populate.sh +++ b/external-dependencies/populate.sh @@ -27,6 +27,7 @@ networkx \ packaging \ pyqtgraph \ python-editor \ +threedi-mi-utils \ threedi-modelchecker \ threedi-schema \ threedidepth \ diff --git a/gui/threedi_plugin_grid_result_dialog.py b/gui/threedi_plugin_grid_result_dialog.py index 3e390a99..08ea4c9b 100644 --- a/gui/threedi_plugin_grid_result_dialog.py +++ b/gui/threedi_plugin_grid_result_dialog.py @@ -9,7 +9,7 @@ from qgis.PyQt.QtWidgets import QAbstractItemView from qgis.core import QgsSettings from qgis.PyQt.QtGui import QStandardItemModel, QStandardItem -from threedi_results_analysis.utils.workingdir import list_local_schematisations +from threedi_mi_utils import list_local_schematisations logger = logging.getLogger(__name__) @@ -199,10 +199,11 @@ def refresh(self): self.messageLabel.setText("Please set your 3Di working directory in the 3Di Models & Simulations settings to be able to load computational grids and results from your 3Di working directory.") return - local_schematisations = list_local_schematisations(threedi_working_dir) + local_schematisations = list_local_schematisations(threedi_working_dir, use_config_for_revisions=False) for schematisation_id, local_schematisation in local_schematisations.items(): # Iterate over revisions for revision_number, local_revision in local_schematisation.revisions.items(): + num_of_results = len(local_revision.results_dirs) # Iterate over results for result_dir in local_revision.results_dirs: schema_item = QStandardItem(local_schematisation.name) @@ -218,7 +219,7 @@ def refresh(self): self.model.appendRow([schema_item, revision_item, result_item]) # In case no results are present, but a gridadmin is present, we still add the grid, but without result item - if len(local_revision.results_dirs) == 0 and os.path.exists(os.path.join(local_revision.grid_dir, "gridadmin.h5")): + if num_of_results == 0 and os.path.exists(os.path.join(local_revision.grid_dir, "gridadmin.h5")): schema_item = QStandardItem(local_schematisation.name) schema_item.setEditable(False) revision_item = QStandardItem(str(revision_number)) diff --git a/utils/workingdir.py b/utils/workingdir.py deleted file mode 100644 index 9ce429f9..00000000 --- a/utils/workingdir.py +++ /dev/null @@ -1,227 +0,0 @@ -from collections import OrderedDict -from itertools import chain -import json -import re -import os -from threedi_results_analysis.utils.utils import listdirs - - -class LocalRevision: - """Local revision directory structure representation.""" - - def __init__(self, local_schematisation, revision_number): - self.local_schematisation = local_schematisation - self.number = revision_number - - def structure_is_valid(self): - """Check if all revision subpaths are present.""" - is_valid = all(os.path.exists(p) if p else False for p in self.subpaths) - return is_valid - - @property - def sub_dir(self): - """Get schematisation revision subdirectory name.""" - subdirectory = f"revision {self.number}" - return subdirectory - - @property - def main_dir(self): - """Get schematisation revision main directory path.""" - schematisation_dir_path = self.local_schematisation.main_dir - schematisation_revision_dir_path = os.path.join(schematisation_dir_path, self.sub_dir) - return schematisation_revision_dir_path - - @property - def admin_dir(self): - """Get schematisation revision admin directory path.""" - admin_dir_path = os.path.join(self.main_dir, "admin") - return admin_dir_path - - @property - def grid_dir(self): - """Get schematisation revision grid directory path.""" - grid_dir_path = os.path.join(self.main_dir, "grid") - return grid_dir_path - - @property - def results_dir(self): - """Get schematisation revision results directory path.""" - grid_dir_path = os.path.join(self.main_dir, "results") - return grid_dir_path - - @property - def results_dirs(self): - """Get all (full) result folders""" - if not os.path.isdir(self.results_dir): - return [] - return listdirs(self.results_dir) - - @property - def schematisation_dir(self): - """Get schematisation revision schematisation directory path.""" - grid_dir_path = os.path.join(self.main_dir, "schematisation") - return grid_dir_path - - @property - def raster_dir(self): - """Get schematisation revision raster directory path.""" - rasters_dir_path = os.path.join(self.main_dir, "schematisation", "rasters") - return rasters_dir_path - - @property - def subpaths(self): - """Revision directory sub-paths.""" - paths = [ - self.admin_dir, - self.grid_dir, - self.results_dir, - self.schematisation_dir, - self.raster_dir, - ] - return paths - - -class WIPRevision(LocalRevision): - """Local Work In Progress directory structure representation.""" - - @property - def sub_dir(self): - """Get schematisation revision subdirectory name.""" - subdirectory = "work in progress" - return subdirectory - - @property - def main_dir(self): - """Get schematisation revision main directory path.""" - schematisation_dir_path = self.local_schematisation.main_dir - schematisation_revision_dir_path = os.path.join(schematisation_dir_path, self.sub_dir) - return schematisation_revision_dir_path - - @property - def admin_dir(self): - """Get schematisation revision admin directory path.""" - admin_dir_path = os.path.join(self.main_dir, "admin") - return admin_dir_path - - @property - def grid_dir(self): - """Get schematisation revision grid directory path.""" - grid_dir_path = os.path.join(self.main_dir, "grid") - return grid_dir_path - - @property - def results_dir(self): - """Get schematisation revision results directory path.""" - grid_dir_path = os.path.join(self.main_dir, "results") - return grid_dir_path - - @property - def schematisation_dir(self): - """Get schematisation revision schematisation directory path.""" - grid_dir_path = os.path.join(self.main_dir, "schematisation") - return grid_dir_path - - @property - def raster_dir(self): - """Get schematisation revision raster directory path.""" - rasters_dir_path = os.path.join(self.main_dir, "schematisation", "rasters") - return rasters_dir_path - - -class LocalSchematisation: - """Local revision directory structure representation.""" - - def __init__(self, working_dir, schematisation_pk, schematisation_name, parent_revision_number=None): - self.working_directory = working_dir - self.id = schematisation_pk - self.name = schematisation_name - self.revisions = OrderedDict() - self.wip_revision = WIPRevision(self, parent_revision_number) if parent_revision_number is not None else None - - @classmethod - def initialize_from_location(cls, schematisation_dir, use_config_for_revisions=True): - """ - Initialize local schematisation structure from the root schematisation dir. - In case use_config_for_revisions is True, the revisions are derived from the json file, - otherwise the schematisation dir is scanned for "revision" folders. - """ - working_dir = os.path.dirname(schematisation_dir) - if not os.path.isdir(schematisation_dir): - return None - config_path = os.path.join(schematisation_dir, "admin", "schematisation.json") - schema_metadata = cls.read_schematisation_metadata(config_path) - fallback_id = fallback_name = os.path.basename(schematisation_dir) - schematisation_pk = schema_metadata.get("id", fallback_id) - schematisation_name = schema_metadata.get("name", fallback_name) - local_schematisation = cls(working_dir, schematisation_pk, schematisation_name) - - if use_config_for_revisions: - revision_numbers = schema_metadata.get("revisions", []) - else: - folders = [ - os.path.basename(d) - for d in listdirs(schematisation_dir) - if os.path.basename(d).startswith("revision") - ] - revision_numbers = [int(re.findall(r'^revision (\d+)', folder)[0]) for folder in folders] - - for revision_number in revision_numbers: - local_revision = LocalRevision(local_schematisation, revision_number) - local_schematisation.revisions[revision_number] = local_revision - - wip_parent_revision_number = schema_metadata.get("wip_parent_revision") - if wip_parent_revision_number is not None: - local_schematisation.wip_revision = WIPRevision(local_schematisation, wip_parent_revision_number) - - return local_schematisation - - @staticmethod - def read_schematisation_metadata(schematisation_config_path): - """Read schematisation metadata from the JSON file.""" - if not os.path.exists(schematisation_config_path): - return {} - with open(schematisation_config_path, "r+") as config_file: - return json.load(config_file) - - def structure_is_valid(self): - """Check if all schematisation subpaths are present.""" - subpaths_collections = [self.subpaths] - subpaths_collections += [local_revision.subpaths for local_revision in self.revisions.values()] - subpaths_collections.append(self.wip_revision.subpaths) - is_valid = all(os.path.exists(p) if p else False for p in chain.from_iterable(subpaths_collections)) - return is_valid - - @property - def main_dir(self): - """Get schematisation main directory.""" - schematisation_dir_path = os.path.normpath(os.path.join(self.working_directory, self.name)) - return schematisation_dir_path - - @property - def admin_dir(self): - """Get schematisation admin directory path.""" - admin_dir_path = os.path.join(self.main_dir, "admin") - return admin_dir_path - - @property - def subpaths(self): - """Get schematisation directory sub-paths.""" - paths = [self.admin_dir] - return paths - - @property - def schematisation_config_path(self): - """Get schematisation configuration filepath.""" - config_path = os.path.join(self.admin_dir, "schematisation.json") - return config_path - - -def list_local_schematisations(working_dir): - """Get local schematisations present in the given directory.""" - local_schematisations = OrderedDict() - for basename in os.listdir(working_dir): - full_path = os.path.join(working_dir, basename) - local_schematisation = LocalSchematisation.initialize_from_location(full_path, False) - if local_schematisation is not None: - local_schematisations[local_schematisation.id] = local_schematisation - return local_schematisations