Incosistent rekenresultaat debug

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/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