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