Rewrite TUI als workflow incl. State en Unit Tests

geoprob_pipe/app_object.py

@@ -21,7 +21,7 @@
 from geoprob_pipe.results import Results
 from geoprob_pipe.spatial import Spatial
 from geoprob_pipe.visualizations import Visualizations
-from geoprob_pipe.calculations.systems.build_and_run import build_and_run_system_calculations
+from geoprob_pipe.calculations.systems.build_and_run import BuildAndRunCalculations
 from geoprob_pipe.utils.update_metadata import update_metadata
 import logging
 if TYPE_CHECKING:
@@ -57,7 +57,7 @@ def __init__(self, app_settings: ApplicationSettings) -> None:
         # self._read_calculation_settings()  # TODO: Not part of new version
         # TODO: Unsure if the single statement belongs here. Wouldn't it be part of input data?
 
-        self.calc_results: List[CalcResult] = build_and_run_system_calculations(self)
+        self.calc_results: List[CalcResult] = BuildAndRunCalculations(self).run()
         self.results = Results(self)
 
         # Log finish

geoprob_pipe/calculations/systems/base_objects/system_calculation.py

@@ -14,17 +14,9 @@
     "variation_coefficient": 0.02,
     "maximum_iterations": 1000,
     "relaxation_factor": 0.4,
+    "reuse_calculations": False,
 }
 
-# DEFAULT_RELIABILITY_SETTINGS: Dict[str, Union[str, float, int]] = {
-#     "reliability_method": ReliabilityMethod.form.__str__(),
-#     "variation_coefficient": 0.02,
-#     "maximum_iterations": 10_000,
-#     "relaxation_factor": 0.4,
-#     "epsilon_beta": 0.05,
-#     "relaxation_loops": 10,
-# }
-
 
 class SystemSetup:
 

geoprob_pipe/workflow/__init__.py

@@ -0,0 +1,2 @@
+from geoprob_pipe.workflow.steps import steps
+from geoprob_pipe.workflow.state import State

geoprob_pipe/workflow/actions/__init__.py

@@ -0,0 +1 @@
+from geoprob_pipe.workflow.actions.import_hrd_from_hydranl_db import ActionImportHRDLocationsFromHydraNLDatabase

geoprob_pipe/workflow/actions/import_hrd_from_hydranl_db.py

@@ -0,0 +1,33 @@
+from geoprob_pipe.workflow.base_objects import Action
+from geoprob_pipe.workflow.questions import QuestionImportHRD, QuestionDirHydraNLDatabase
+from geoprob_pipe.cmd_app.spatial_layers.hrd.import_from_hrd import hrd_file_path
+import sqlite3
+from pandas import read_sql
+from geopandas import GeoDataFrame, points_from_xy
+
+
+class ActionImportHRDLocationsFromHydraNLDatabase(Action):
+
+    def execute(self):
+        dir_to_db = self.state.question_answer.retrieve(question=QuestionDirHydraNLDatabase.label)
+        path_to_db = hrd_file_path(hrd_dir=dir_to_db)
+        conn = sqlite3.connect(path_to_db)
+        df = read_sql("SELECT Name, XCoordinate, YCoordinate FROM HRDLocations", conn)
+        gdf = GeoDataFrame(df, geometry=points_from_xy(df["XCoordinate"], df["YCoordinate"]), crs="EPSG:28992")
+        conn.close()
+        gdf = gdf.drop(columns=["XCoordinate", "YCoordinate"])
+        gdf = gdf.rename(columns={"Name": "location_name"})
+        self.state.gdf.store(gdf=gdf, layer_name="hrd_locations")
+
+    @property
+    def should_run(self) -> bool:
+        if not (self.state.question_answer.retrieve(QuestionImportHRD.label) == "Hydra-NL database" and
+                self.state.question_answer.retrieve(QuestionDirHydraNLDatabase.label) is not None):
+            return False
+        return True
+
+    @property
+    def completed(self) -> bool:
+        if self.state.gdf.hrd_locations is None:
+            return False
+        return True

geoprob_pipe/workflow/base_objects.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+from geoprob_pipe.utils.validation_messages import BColors
+from dataclasses import dataclass
+if TYPE_CHECKING:
+    from geoprob_pipe.workflow.state import State
+
+
+class ClassName:
+    def __get__(self, obj, cls):
+        return cls.__name__
+
+
+class Step(ABC):
+    label: str = ClassName()
+
+    def __init__(self, state: State):
+        self.state = state
+
+    @property
+    @abstractmethod
+    def should_run(self) -> bool:
+        """ Return True if this step should be run, based on the state of the project. """
+        raise NotImplementedError()
+
+    @property
+    @abstractmethod
+    def completed(self) -> bool:
+        """ Return True if this step is already completed. Otherwise, step must be run. """
+        raise NotImplementedError()
+
+    @abstractmethod
+    def execute(self):
+        raise NotImplementedError()
+
+
+@dataclass
+class ValidationResult:
+    is_valid: bool
+    message: str | None = None
+    manipulated_answer: str | None = None
+
+
+class Question(Step):
+
+    @abstractmethod
+    def ask(self):
+        raise NotImplementedError()
+
+    def execute(self):
+        answer = self.ask_until_valid()
+        self.state.question_answer.store(question_label=self.label, answer=answer)
+
+    @staticmethod
+    @abstractmethod
+    def validate(answer) -> ValidationResult:
+        raise NotImplementedError()
+
+    def ask_until_valid(self):
+        while True:
+            answer = self.ask()
+
+            # Validate
+            result: ValidationResult = self.validate(answer)
+
+            # Is valid
+            if result.is_valid:
+                if result.manipulated_answer:
+                    return result.manipulated_answer
+                return answer
+
+            # Is not valid
+            print(f"{BColors.WARNING}{result.message}{BColors.ENDC}")
+
+
+class Action(Step):
+
+    @abstractmethod
+    def execute(self):
+        raise NotImplementedError()

geoprob_pipe/workflow/cmd.py

@@ -0,0 +1,22 @@
+from geoprob_pipe.workflow import steps, State
+import typer
+
+
+state = State(file_path=r"path\to\tmp.gpkg")
+
+app = typer.Typer(help="GeoProb-Pipe - CLI applicatie voor probabilistische piping berekeningen.", add_completion=False)
+
+
+@app.callback(invoke_without_command=True)
+def main(ctx: typer.Context):
+    """ Default entry point for `geoprob-pipe`. Runs when no subcommand is specified. """
+    if ctx.invoked_subcommand is None:
+        for obj in steps:
+            step = obj(state=state)
+            print(f"{step.label}: {step.should_run=} {step.completed=}")
+            if step.should_run and not step.completed:
+                step.execute()
+
+
+if __name__ == "__main__":
+    app()

geoprob_pipe/workflow/questions/__init__.py

@@ -0,0 +1,2 @@
+from geoprob_pipe.workflow.questions.import_hrd import QuestionImportHRD
+from geoprob_pipe.workflow.questions.dir_hydranl_db import QuestionDirHydraNLDatabase

geoprob_pipe/workflow/questions/dir_hydranl_db.py

@@ -0,0 +1,57 @@
+from geoprob_pipe.workflow.base_objects import Question, ValidationResult
+from geoprob_pipe.workflow.questions import QuestionImportHRD
+from typing import Optional
+from InquirerPy import inquirer
+import os
+
+
+def _folder_contains_hrd_db(hrd_dir: str) -> bool:
+    cnt_sql_files = 0
+    cnt_config_files = 0
+    cnt_hlcd_files = 0
+
+    for file in os.listdir(hrd_dir):
+        filename = os.fsdecode(file)
+        if filename.endswith(".sqlite"):
+            cnt_sql_files += 1
+        if filename.endswith(".config.sqlite"):
+            cnt_config_files += 1
+        if filename.endswith("hlcd.sqlite"):
+            cnt_hlcd_files += 1
+
+    if cnt_sql_files == 3 and cnt_config_files == 1 and cnt_hlcd_files == 1:
+        return True
+    return False
+
+class QuestionDirHydraNLDatabase(Question):
+
+    def ask(self) -> str:
+        return inquirer.text(
+            message="Specificeer het volledige pad naar de bestandsmap met de Hydra-NL database. "
+                    "Dat zijn de hlcd, config en het database .sqlite-bestand zelf.",
+        ).execute()
+
+    @staticmethod
+    def validate(answer) -> ValidationResult:
+        manipulated_answer = answer.replace('"', '')
+        if not os.path.isdir(manipulated_answer):
+            return ValidationResult(False, "The provided path is not a directory.")
+        if not _folder_contains_hrd_db(hrd_dir=manipulated_answer):
+            return ValidationResult(
+                False, "The provided directory does not contain the necessary Hydra-NL database-files.")
+        # TODO: Validate HRD-point locations are valid
+        return ValidationResult(True, manipulated_answer=manipulated_answer)
+
+    @property
+    def should_run(self) -> bool:
+        answer: Optional[str] = self.state.question_answer.retrieve(QuestionImportHRD.label)
+        if answer != "Hydra-NL database":
+            return False
+        return True
+
+    @property
+    def completed(self) -> bool:
+        answer: Optional[str] = self.state.question_answer.retrieve(self.label)
+        if answer is None:
+            return False
+        return True

geoprob_pipe/workflow/questions/import_hrd.py

@@ -0,0 +1,39 @@
+from geoprob_pipe.workflow.base_objects import Question, ValidationResult
+from typing import Optional
+from InquirerPy import inquirer
+
+
+CHOICES = [
+    "Hydra-NL database",
+    "Ander GeoProb-Pipe bestand",
+    "Nee",
+]
+
+
+class QuestionImportHRD(Question):
+
+    def ask(self) -> str:
+        return inquirer.select(
+            message="Wil je overschrijdingsfrequentielijnen importeren? En zo ja, uit welke bron? "
+                    "Je kunt op een later moment handmatig overschrijdingsfrequentielijnen toevoegen aan het "
+                    "bestand met invoer-Excel.",
+            choices=CHOICES,
+            default=CHOICES[0],
+        ).execute()
+
+    @staticmethod
+    def validate(answer):
+        if answer not in CHOICES:
+            return ValidationResult(False, f"Kies één van de volgende opties: {CHOICES}.")
+        return ValidationResult(True)
+
+    @property
+    def should_run(self) -> bool:
+        return True  # Always ask if HRD-data should be imported.
+
+    @property
+    def completed(self) -> bool:
+        answer: Optional[str] = self.state.question_answer.retrieve(self.label)
+        if answer is None or answer == CHOICES[2]:
+            return False
+        return True