diff --git a/geoprob_pipe/app_object.py b/geoprob_pipe/app_object.py index 549c6692..32716b57 100644 --- a/geoprob_pipe/app_object.py +++ b/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 diff --git a/geoprob_pipe/calculations/systems/base_objects/system_calculation.py b/geoprob_pipe/calculations/systems/base_objects/system_calculation.py index 112135d0..73618798 100644 --- a/geoprob_pipe/calculations/systems/base_objects/system_calculation.py +++ b/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: diff --git a/geoprob_pipe/calculations/systems/build_and_run.py b/geoprob_pipe/calculations/systems/build_and_run.py index 085970c9..e46354a5 100644 --- a/geoprob_pipe/calculations/systems/build_and_run.py +++ b/geoprob_pipe/calculations/systems/build_and_run.py @@ -17,9 +17,9 @@ collect_stochast_values, calculate_derived_values) import logging from pandas import DataFrame - +import os if TYPE_CHECKING: - from geoprob_pipe import GeoProbPipe + from geoprob_pipe import GeoProbPipe, SystemCalculation from geoprob_pipe.calculations.systems.base_objects\ .base_system_build import BaseSystemBuilder from geoprob_pipe.utils.validation_messages import ValidationMessages @@ -70,13 +70,7 @@ def _init_worker( to_run_vakken_ids=to_run_vakken_ids)) -def _worker(row_unique: dict): - """ De worker functie die op de parallelle rekenkernen wordt gedraaid. - - :param row_unique: Identificatie naar unieke berekening, bijvoorbeeld - {'uittredepunt_id': 1, 'ondergrondscenario_naam': 'scenario1', 'vak_id': 4}. - :return: Tuple[Optional[CalcResult], Optional[str], Optional[dict]] - """ +def _logging_code(): log_buffer = StringIO() buffer_handler = logging.StreamHandler(log_buffer) buffer_handler.setLevel(logging.DEBUG) @@ -92,36 +86,91 @@ def _worker(row_unique: dict): root.addHandler(buffer_handler) logging.captureWarnings(True) + return log_buffer, buffer_handler, root, prev_level + + +def _run_calculation(row_unique: dict) -> SystemCalculation: + debug: bool = os.environ.get("GEOPROB_DEBUG", False) + if debug: logger.debug("Start berekening voor %s", row_unique) + + # Run calculation + calc: SystemCalculation = _BUILDER.build_instance(row_unique=row_unique) + calc.run() + if debug: logger.debug("SystemCalculation voltooid.") + + # Remainder is logging in case of validation messages or debug modus + if calc.validation_messages.df: logger.debug(f"Validation messages:\n{calc.validation_messages.df}") + + if debug: + logger.debug("Limit states print:") + for lm in calc.results.dps_limit_states: + lm.print() + + logger.debug("Combine project print:") + calc.results.combine_project.design_point.print() + + logger.debug("Reliability project print:") + calc.results.reliability_project.design_point.print() + + return calc + + +def _collect_results(calc: SystemCalculation) -> CalcResult: + debug: bool = os.environ.get("GEOPROB_DEBUG", False) + df_limit_state = collect_df_beta_limit_state(calc) + if debug: logger.debug(f"df_limit_state:\n{df_limit_state}") + if any(r == 0 for r in df_limit_state.total_model_runs): + logger.warning("Limit state with 0 total model runs encountered. Notify developer and re-run calculations. ") + + df_scenario_rp = collect_df_beta_scenario_rp(calc) + if debug: logger.debug(f"df_scenario_rp:\n{df_scenario_rp}") + + df_scenario_cp = collect_df_beta_scenario_cp(calc) + if debug: logger.debug(f"df_scenario_cp:\n{df_scenario_cp}") + + df_scenario_final = collect_df_beta_scenario_final(calc) + + df_stochast = collect_stochast_values(calc, df_scenario_final=df_scenario_final) + if debug: logger.debug(f"df_stochast:\n{df_stochast.to_string()}") + + if df_scenario_cp.converged is True and any(a >= 0.99 for a in df_stochast.alpha): + logger.warning("Unrealistically dominant (alpha >= 0.99) parameter found in combined project.") + if df_scenario_rp.converged is True and any(a >= 0.99 for a in df_stochast.alpha): + logger.warning("Unrealistically dominant (alpha >= 0.99) parameter found in reliability project.") + df_derived = calculate_derived_values(df_scenarios_final=df_scenario_final, geohydrologisch_model=_MODEL) + df_scenario_rp = df_scenario_rp.drop(columns=["system_calculation"]) + df_scenario_cp = df_scenario_cp.drop(columns=["system_calculation"]) + df_scenario_final = df_scenario_final.drop(columns=["system_calculation"]) + + return CalcResult( + df_limit_state=df_limit_state, df_scenario_rp=df_scenario_rp, df_scenario_cp=df_scenario_cp, + df_scenario_final=df_scenario_final, df_stochast=df_stochast, df_derived=df_derived, + validation_message=calc.validation_messages) + + +def _worker(row_unique: dict): + """ De worker functie die op de parallelle rekenkernen wordt gedraaid. + + :param row_unique: Identificatie naar unieke berekening, bijvoorbeeld + {'uittredepunt_id': 1, 'ondergrondscenario_naam': 'scenario1', 'vak_id': 4}. + :return: Tuple[Optional[CalcResult], Optional[str], Optional[dict]] + """ + log_buffer, buffer_handler, root, prev_level = _logging_code() + # noinspection PyBroadException try: with redirect_stdout(log_buffer), redirect_stderr(log_buffer): - # Build and run calculations - calc = _BUILDER.build_instance(row_unique=row_unique) - calc.run() - - # Collect results - df_limit_state = collect_df_beta_limit_state(calc) - df_scenario_rp = collect_df_beta_scenario_rp(calc) - df_scenario_cp = collect_df_beta_scenario_cp(calc) - df_scenario_final = collect_df_beta_scenario_final(calc) - df_stochast = collect_stochast_values(calc, df_scenario_final=df_scenario_final) - df_derived = calculate_derived_values(df_scenarios_final=df_scenario_final, geohydrologisch_model=_MODEL) - df_scenario_rp = df_scenario_rp.drop(columns=["system_calculation"]) - df_scenario_cp = df_scenario_cp.drop(columns=["system_calculation"]) - df_scenario_final = df_scenario_final.drop(columns=["system_calculation"]) - - # Return results (without calculation object) - return CalcResult( - df_limit_state=df_limit_state, df_scenario_rp=df_scenario_rp, df_scenario_cp=df_scenario_cp, - df_scenario_final=df_scenario_final, df_stochast=df_stochast, df_derived=df_derived, - validation_message=calc.validation_messages - ), None, None + # logger = logging.getLogger(__name__) + calc = _run_calculation(row_unique) + result = _collect_results(calc) + return result, log_buffer.getvalue(), row_unique except Exception: tb = traceback.format_exc() log_buffer.write(tb) buffer_handler.flush() return None, log_buffer.getvalue(), row_unique + finally: # Handler altijd verwijderen logging.captureWarnings(False) @@ -130,92 +179,120 @@ def _worker(row_unique: dict): buffer_handler.close() -def build_and_run_system_calculations(geoprob_pipe: GeoProbPipe) -> List[CalcResult]: - """ In deze functie worden de parameters voor de berekeningen verzamelt, - aan de workers gegeven en vervolgens de resultaten verzameld. - """ - geohydrologisch_model = geoprob_pipe.input_data.geohydrologisch_model - geopackage_filepath = ( - geoprob_pipe.input_data.app_settings.geopackage_filepath) - to_run_vakken_ids = geoprob_pipe.input_data.app_settings.to_run_vakken_ids - system_builder: BaseSystemBuilder = ( - CALCULATION_MAPPER[geohydrologisch_model]['system_builder']( - geopackage_filepath=geopackage_filepath, - to_run_vakken_ids=to_run_vakken_ids)) +class BuildAndRunCalculations: + """ In dit object worden de parameters voor de berekeningen verzamelt, aan de workers gegeven en vervolgens de + resultaten verzameld. """ - logger.info("Now building and running calculations...") - df_unique_combos = system_builder.setup_iteration_df() - # Bepaal de parameters voor de multiprocessing setup en de logger - n_threads: int = cpu_count() - 1 - n_calc_totaal: int = len(df_unique_combos) - # Minimaal 5 berekeningen per chunk en grootte van chunk beperken - # zodat er gelogd kan worden. - chunk_size: int = max(math.ceil(n_calc_totaal / (n_threads * 10)), 5) - logger.info( - f"Running {n_calc_totaal} calculations in chunks of {chunk_size}" - f" with {n_threads} parallel threads.") - char_len_total = str(n_calc_totaal).__len__() - logger.info( - f"Progress: {0:>{char_len_total}} / {n_calc_totaal} calculations.") - - # Dicts zijn gemakkelijker te pickelen en daardoor sneller te - # verwerken dan pandas series. - rows = [dict(zip(df_unique_combos.columns, r)) - for r in df_unique_combos.itertuples(index=False, name=None)] - - last_report = time.time() - done = 0 - results: List[CalcResult] = [] - pool_size = max(min(math.floor(n_calc_totaal / chunk_size), n_threads), 1) - - # Multiprocessing setup - error_rows = [] - with Pool(processes=pool_size, initializer=_init_worker, initargs=( - geohydrologisch_model, geopackage_filepath, to_run_vakken_ids)) as pool: - - for res, error_logs, row in pool.imap_unordered(_worker, rows, chunksize=chunk_size): - if isinstance(res, CalcResult): - results.append(res) - if isinstance(error_logs, str): - error_rows.append({ - "uittredepunt_id": row["uittredepunt_id"], - "ondergrondscenario_naam": row["ondergrondscenario_naam"], - "vak_id": row["vak_id"], - "error_logs": error_logs, - }) - done += 1 - - # Alleen kijken of er gelogd moet worden bij de laatste - # berekening die uit de chunk komt. - if done % chunk_size != 0: - continue - - # Alleen loggen wanneer 30 seconden is gepasseerd - now = time.time() - if now - last_report < 30.0: - continue - - # Log - error_count_append = "" - if error_rows.__len__() > 0: - error_count_append = f" (of which {error_rows.__len__()} failed calculations)" - logger.info(f"Progress: {done:>{char_len_total}} / {n_calc_totaal} calculations{error_count_append}.") - last_report = now - - # Push errors to database (if any) - conn = sqlite3.connect(geoprob_pipe.input_data.app_settings.geopackage_filepath) - table_name = "calculation_error_logs" - if error_rows.__len__() > 0: - df_errors = DataFrame(data=error_rows) - df_errors.to_sql(table_name, conn, if_exists="replace", index=False) - conn.close() - logger.error(f"There are {error_rows.__len__()} failed calculations. Error logs are stored inside the " - f"GeoPacakge in table '{table_name}'.") - else: - # Remove old table (if exists) - cur = conn.cursor() - cur.execute(f"DROP TABLE IF EXISTS {table_name};") + def __init__(self, geoprob_pipe: GeoProbPipe): + """ Init zet dit object op, maar het uitvoeren van de berekeningen gaat via de method .run(). """ + + self.geoprob_pipe: GeoProbPipe = geoprob_pipe + self.geohydrologisch_model: str = geoprob_pipe.input_data.geohydrologisch_model + self.geopackage_filepath: str = geoprob_pipe.input_data.app_settings.geopackage_filepath + self.to_run_vakken_ids: str = geoprob_pipe.input_data.app_settings.to_run_vakken_ids + + self._construct_system_builder_and_settings() + self._setup_calculation_progress_variables() + + # Run logic with method .run() + + def _construct_system_builder_and_settings(self): + """ Opzetten van de system builder en andere settings voor de berekeningen. """ + + logger.info("Now preparing for calculations...") + self.system_builder: BaseSystemBuilder = CALCULATION_MAPPER[self.geohydrologisch_model]['system_builder']( + geopackage_filepath=self.geopackage_filepath, + to_run_vakken_ids=self.to_run_vakken_ids) + self.df_unique_combos = self.system_builder.setup_iteration_df() + + # Bepaal de parameters voor de multiprocessing setup en de logger + self.n_threads: int = cpu_count() - 1 + self.n_calc_totaal: int = len(self.df_unique_combos) + + # Minimaal 5 berekeningen per chunk en grootte van chunk beperken zodat er gelogd kan worden. + self.chunk_size: int = max(math.ceil(self.n_calc_totaal / (self.n_threads * 10)), 5) + + def _setup_calculation_progress_variables(self): + """ Simpel initiëren van een aantal variabelen die nodig zijn tijdens de berekeningen. """ + + self.last_report: float = time.time() + self.done = 0 + self.log_errors = 0 + self.error_rows = [] + self.char_len_total: int = str(self.n_calc_totaal).__len__() + self.log_rows = [] + + def _report_calculation_progress_to_user(self): + """ Gedurende het uitvoeren van de berekening koppelt deze method terug wat de progressie is. """ + + # If finished + error_count_append = "" + if self.log_errors > 0: + error_count_append = f" (of which {self.log_errors} failed calculations)" + if self.n_calc_totaal == self.done: + logger.info(f"Progress: {self.done:>{self.char_len_total}} / {self.n_calc_totaal} calculations" + f"{error_count_append}.") + + # Alleen kijken of er gelogd moet worden bij de laatste berekening die uit de chunk komt. + if self.done % self.chunk_size != 0: + return + + # Alleen loggen wanneer 30 seconden is gepasseerd + now = time.time() + if now - self.last_report < 30.0: + return + + logger.info( + f"Progress: {self.done:>{self.char_len_total}} / {self.n_calc_totaal} calculations{error_count_append}.") + self.last_report = now + + def _push_resulting_error_messages_to_database(self): + """ Aan eind van run pushed deze method de errors (if any) naar de database. """ + + conn = sqlite3.connect(self.geopackage_filepath) + table_name = "calculation_logs" + df_logs = DataFrame(data=self.log_rows) + df_logs.to_sql(table_name, conn, if_exists="replace", index=False) conn.commit() conn.close() + if self.log_errors > 0: + logger.error(f"There are {self.log_errors} failed calculations. The calculation logs are stored inside the " + f"GeoPackage in table '{table_name}'. The following rows are marked:" + f"\n{self.error_rows}") + + def run(self) -> List[CalcResult]: + + logger.info( + f"Running {self.n_calc_totaal} calculations in chunks of {self.chunk_size}" + f" with {self.n_threads} parallel threads.") + logger.info( + f"Progress: {0:>{self.char_len_total}} / {self.n_calc_totaal} calculations.") + + rows = [dict(zip(self.df_unique_combos.columns, r)) + for r in self.df_unique_combos.itertuples(index=False, name=None)] + pool_size = max(min(math.floor(self.n_calc_totaal / self.chunk_size), self.n_threads), 1) + + # Multiprocessing setup + results: List[CalcResult] = [] + with Pool(processes=pool_size, initializer=_init_worker, initargs=( + self.geohydrologisch_model, self.geopackage_filepath, self.to_run_vakken_ids)) as pool: + + for res, logs, row in pool.imap_unordered(_worker, rows, chunksize=self.chunk_size): + if isinstance(res, CalcResult): + results.append(res) + if isinstance(logs, str): + self.log_rows.append({ + "uittredepunt_id": row["uittredepunt_id"], + "ondergrondscenario_naam": row["ondergrondscenario_naam"], + "vak_id": row["vak_id"], + "logs": logs, + }) + if any(level in logs for level in ("WARNING", "ERROR", "CRITICAL")): + self.log_errors += 1 + self.error_rows.append(row) + self.done += 1 + + self._report_calculation_progress_to_user() - return results + self._push_resulting_error_messages_to_database() + return results diff --git a/geoprob_pipe/workflow/__init__.py b/geoprob_pipe/workflow/__init__.py new file mode 100644 index 00000000..59ab628d --- /dev/null +++ b/geoprob_pipe/workflow/__init__.py @@ -0,0 +1,2 @@ +from geoprob_pipe.workflow.steps import steps +from geoprob_pipe.workflow.state import State diff --git a/geoprob_pipe/workflow/actions/__init__.py b/geoprob_pipe/workflow/actions/__init__.py new file mode 100644 index 00000000..3e2a1a92 --- /dev/null +++ b/geoprob_pipe/workflow/actions/__init__.py @@ -0,0 +1 @@ +from geoprob_pipe.workflow.actions.import_hrd_from_hydranl_db import ActionImportHRDLocationsFromHydraNLDatabase \ No newline at end of file diff --git a/geoprob_pipe/workflow/actions/import_hrd_from_hydranl_db.py b/geoprob_pipe/workflow/actions/import_hrd_from_hydranl_db.py new file mode 100644 index 00000000..9ccd8c00 --- /dev/null +++ b/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 diff --git a/geoprob_pipe/workflow/base_objects.py b/geoprob_pipe/workflow/base_objects.py new file mode 100644 index 00000000..ddf8a037 --- /dev/null +++ b/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() diff --git a/geoprob_pipe/workflow/cmd.py b/geoprob_pipe/workflow/cmd.py new file mode 100644 index 00000000..8478a976 --- /dev/null +++ b/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() diff --git a/geoprob_pipe/workflow/questions/__init__.py b/geoprob_pipe/workflow/questions/__init__.py new file mode 100644 index 00000000..cde1384a --- /dev/null +++ b/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 diff --git a/geoprob_pipe/workflow/questions/dir_hydranl_db.py b/geoprob_pipe/workflow/questions/dir_hydranl_db.py new file mode 100644 index 00000000..95051e32 --- /dev/null +++ b/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 diff --git a/geoprob_pipe/workflow/questions/import_hrd.py b/geoprob_pipe/workflow/questions/import_hrd.py new file mode 100644 index 00000000..ac84648a --- /dev/null +++ b/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 diff --git a/geoprob_pipe/workflow/state.py b/geoprob_pipe/workflow/state.py new file mode 100644 index 00000000..34346307 --- /dev/null +++ b/geoprob_pipe/workflow/state.py @@ -0,0 +1,122 @@ +import os +from typing import Optional, Dict +import sqlite3 +from geopandas import GeoDataFrame, read_file +from pathlib import Path +from uuid import uuid4 +from geoprob_pipe.utils.validation_messages import BColors +import fiona + + +class QuestionAnswer: + """ The terminal user interface has a workflow of questions that the users answers. Based on this, and imported data + GeoProb-Pipe determines the state of the application. The answers to these questions (this part of the state) are + stored in the geopackage in table 'workflow_questions'. """ + + def __init__(self, file_path: str): + self.file_path: str = file_path + + def store(self, question_label: str, answer: str): + """ This method stores the answer to a question. """ + + conn = sqlite3.connect(self.file_path) + cursor = conn.cursor() + + cursor.execute(""" + CREATE TABLE IF NOT EXISTS workflow_questions ( + question_label TEXT PRIMARY KEY, + answer TEXT + ) + """) + + cursor.execute(f""" + INSERT INTO workflow_questions ( + question_label, + answer + ) + VALUES (?, ?) + ON CONFLICT(question_label) + DO UPDATE SET answer = excluded.answer + """, (question_label, answer)) + + conn.commit() + conn.close() + + def retrieve(self, question: str) -> Optional[str]: + """ This method retrieves the answer to a question (if already stored). """ + conn = sqlite3.connect(self.file_path) + cursor = conn.cursor() + try: + cursor.execute(f""" + SELECT answer + FROM workflow_questions + WHERE question_label = '{question}' + LIMIT 1; + """) + except sqlite3.OperationalError: # table does not exist + return None + result = cursor.fetchone() + if not result: + return None + return result[0] + + +class GeoDataFrames: + + def __init__(self, file_path: Optional[str] = None): + self.file_path: str = file_path + self._read_geodataframes: Dict[str, GeoDataFrame] = {} + + def store(self, gdf: GeoDataFrame, layer_name: str): + gdf.to_file(Path(self.file_path), layer=layer_name, driver="GPKG") + print(f"{BColors.OKBLUE}" + f"✅ Geografische tabel '{layer_name}' toegevoegd aan het GeoProb-Pipe GeoPackage.{BColors.ENDC}") + + def retrieve(self, layer_name: str) -> Optional[GeoDataFrame]: + # First check if already retrieved before + if layer_name in self._read_geodataframes.keys(): + return self._read_geodataframes[layer_name] + + # Otherwise retrieve from GeoProb-Pipe-file + layers = fiona.listlayers(self.file_path) + if layer_name not in layers: + return None + return read_file(self.file_path, layer=layer_name) + + @property + def hrd_locations(self) -> Optional[GeoDataFrame]: + return self.retrieve(layer_name="hrd_locations") + + +class DataFrames: + + def __init__(self): + pass + + +def _if_needed_create_dummy_gpkg(file_path: Optional[str] = None, file_dir: Optional[str] = None) -> str: + if (file_path is None and file_dir is None) or (file_path is not None and file_dir is not None): + raise ValueError(f"Specify either the path to the GeoProb-Pipe-file, or a directory where a dummy can " + f"can be created.") + elif file_dir is not None: + import geopandas as gpd + gdf = gpd.GeoDataFrame(geometry=[]) + path_to_gpkg = os.path.join(file_dir, f"dummy_{uuid4().__str__()}.geoprob_pipe.gpkg") + gdf.to_file(Path(path_to_gpkg, driver="GPKG")) + file_path = path_to_gpkg + return file_path + + +class State: + """ Through the terminal user interface the user selects choices/preferences (questions and answers) and imports + data. The state of the application is the state of these answers and imported data. """ + + def __init__(self, file_path: Optional[str] = None, file_dir: Optional[str | Path] = None): + """ + + :param file_path: Path to the GeoProb-Pipe file. + """ + self.file_path: str = _if_needed_create_dummy_gpkg(file_path, file_dir) + self.question_answer = QuestionAnswer(self.file_path) + self.gdf = GeoDataFrames(self.file_path) + self.df = DataFrames() diff --git a/geoprob_pipe/workflow/steps.py b/geoprob_pipe/workflow/steps.py new file mode 100644 index 00000000..d6daee35 --- /dev/null +++ b/geoprob_pipe/workflow/steps.py @@ -0,0 +1,27 @@ +from geoprob_pipe.workflow.base_objects import Step +from typing import List, Type +from geoprob_pipe.workflow.questions import QuestionImportHRD, QuestionDirHydraNLDatabase +from geoprob_pipe.workflow.actions import ActionImportHRDLocationsFromHydraNLDatabase + +steps: List[Type[Step]] = [ + + # HRD + QuestionImportHRD, + QuestionDirHydraNLDatabase, + ActionImportHRDLocationsFromHydraNLDatabase, + # ActionImportTrajectParametersFromHydraNLDatabase, + # QuestionFilePathGeoProbPipeFileWithHRD, + # ActionImportHRDFromOtherGeoProbPipeFile, + # ActionImportTrajectParametersFromOtherGeoProbPipeFile, + + # Traject parameters + # QuestionTrajectID, + # QuestionSignaleringswaarde, + # QuestionOndergrens, + # QuestionW, + # QuestionIsBovenrivierengebied, + + # Uittredepunten + # QuestionPathToUittredepuntenGISFile, + # ActionImportUittredepuntenGISFile, +] diff --git a/repo_utils/utils.py b/repo_utils/utils.py index 20c0aa67..99e32144 100644 --- a/repo_utils/utils.py +++ b/repo_utils/utils.py @@ -3,12 +3,22 @@ from typing import Optional +def find_repo_root(): + """ Alternative version to repository_root_path, which does not always seem to work in PyTest. """ + from pathlib import Path + import subprocess + return Path(subprocess.check_output(["git", "rev-parse", "--show-toplevel"], text=True).strip()) + + def repository_root_path() -> Optional[str]: # Bold guess, it is the execution path base_dir = os.getcwd() + print(f"{base_dir=}") try: + print(f"Now trying Repo-class") repo_root = Repo(base_dir, search_parent_directories=False).working_tree_dir + print(f"From Repo-class {repo_root=}") return repo_root except InvalidGitRepositoryError: pass diff --git a/tests/workflow/__init__.py b/tests/workflow/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/workflow/actions/__init__.py b/tests/workflow/actions/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/workflow/actions/test_import_hrd_from_hydranl_db.py b/tests/workflow/actions/test_import_hrd_from_hydranl_db.py new file mode 100644 index 00000000..db5f5802 --- /dev/null +++ b/tests/workflow/actions/test_import_hrd_from_hydranl_db.py @@ -0,0 +1,21 @@ + + +def test_action_import_hrd_locations_from_hydranl_db(tmp_path): + ## + from geoprob_pipe.workflow import State + from geoprob_pipe.workflow.actions import ActionImportHRDLocationsFromHydraNLDatabase + from geoprob_pipe.workflow.questions import QuestionDirHydraNLDatabase + from repo_utils.utils import find_repo_root + import os + + state = State(file_dir=tmp_path) + repo_root = find_repo_root() + dir_to_db = os.path.join(repo_root, "tests", "systeem_testen", "224", "hrd_files") + state.question_answer.store( + question_label=QuestionDirHydraNLDatabase.label, + answer=dir_to_db) + action = ActionImportHRDLocationsFromHydraNLDatabase(state=state) + action.execute() + assert state.gdf.hrd_locations is not None + assert state.gdf.hrd_locations.__len__() == 13 + ## diff --git a/tests/workflow/questions/__init__.py b/tests/workflow/questions/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/workflow/questions/test_dir_hydranl_db.py b/tests/workflow/questions/test_dir_hydranl_db.py new file mode 100644 index 00000000..b90571bf --- /dev/null +++ b/tests/workflow/questions/test_dir_hydranl_db.py @@ -0,0 +1,22 @@ + + + +def test_question_dir_hydra_nl_database(): + ## + from geoprob_pipe.workflow.questions import QuestionDirHydraNLDatabase + from repo_utils.utils import find_repo_root + import os + repo_root = find_repo_root() + test_answers = [ + os.path.join(repo_root, "tests", "systeem_testen", "224", "hrd_files", "WBI2017_Bovenrijn_224_v04.sqlite"), + os.path.join(repo_root, "tests", "systeem_testen", "224"), + os.path.join(repo_root, "tests", "systeem_testen", "224", "hrd_files"), + ] + test_results = [ + False, + False, + True, + ] + for test_answer, test_result in zip(test_answers, test_results): + assert QuestionDirHydraNLDatabase.validate(answer=test_answer).is_valid == test_result + ## \ No newline at end of file diff --git a/tests/workflow/questions/test_import_hrd.py b/tests/workflow/questions/test_import_hrd.py new file mode 100644 index 00000000..e96b2432 --- /dev/null +++ b/tests/workflow/questions/test_import_hrd.py @@ -0,0 +1,13 @@ + + +def test_question_import_hrd(): + from geoprob_pipe.workflow.questions import QuestionImportHRD + from geoprob_pipe.workflow.questions.import_hrd import CHOICES + import numpy as np + + test_answers = CHOICES.copy() # Force copy, otherwise you change CHOICES with .extend() on the next line. + test_answers.extend(["Ja", None, 1, np.nan]) + test_results = [True, True, True, False, False, False, False] + for test_answer, test_result in zip(test_answers, test_results): + assert QuestionImportHRD.validate(answer=test_answer).is_valid == test_result, \ + f"Failed on {test_answer}, expected {test_result}." diff --git a/tests/workflow/test_workflow.py b/tests/workflow/test_workflow.py new file mode 100644 index 00000000..0f85bc2a --- /dev/null +++ b/tests/workflow/test_workflow.py @@ -0,0 +1,40 @@ + + +def test_workflow(tmp_path): + import os.path + from geoprob_pipe.workflow import State, steps + from geoprob_pipe.workflow.questions import QuestionDirHydraNLDatabase, QuestionImportHRD + from geoprob_pipe.workflow.questions.import_hrd import CHOICES + from geoprob_pipe.workflow.base_objects import Question + from repo_utils.utils import find_repo_root + import random + + repo_root = find_repo_root() + valid_answers = { + QuestionImportHRD.label: CHOICES, + QuestionDirHydraNLDatabase.label: [os.path.join(repo_root, "tests", "systeem_testen", "224", "hrd_files")], + } + # TODO: Hoe specificeer ik een relatief pad wat werkt in Ubuntu GitHub en lokaal bij ons. + # TODO: Sowieso test bestanden opzetten en kijken of ik nog ergens lokale paden gebruik. + + state = State(file_dir=tmp_path) + for obj in steps: + step = obj(state=state) + print(f"\n{step.label}: {step.should_run=} {step.completed=}") + + # Check + if not step.should_run: + continue + if step.completed: + continue + + # Fake 'execute' question + if issubclass(obj, Question): + answer = random.choice(valid_answers[obj.label]) + print(f"Now popping '{answer}' to '{obj.label}'.") + state.question_answer.store(question_label=obj.label, answer=answer) + assert step.completed == True + continue + + # Thus Action + step.execute()