Source code for ogstools.logparser.regexes

# Copyright (c) 2012-2025, OpenGeoSys Community (http://www.opengeosys.org)
#            Distributed under a Modified BSD License.
#            See accompanying file LICENSE.txt or
#            http://www.opengeosys.org/project/license
#

from dataclasses import dataclass
from enum import Enum


[docs] @dataclass class Log: type: str line: int
[docs] @staticmethod def type_str() -> str: return "Log"
[docs] @staticmethod def context_filter() -> list[str]: return []
[docs] class Info(Log):
[docs] @staticmethod def type_str() -> str: return "Info"
[docs] class Termination:
[docs] @staticmethod def context_filter() -> list[str]: return []
[docs] class WarningType(Log):
[docs] @staticmethod def type_str() -> str: return "Warning"
[docs] class ErrorType(Log):
[docs] @staticmethod def type_str() -> str: return "Error"
[docs] class CriticalType(Log):
[docs] @staticmethod def type_str() -> str: return "Critical"
[docs] @dataclass class MPIProcess(Info): mpi_process: int
[docs] @dataclass class NoRankOutput: pass
[docs] @dataclass class OGSVersionLog(MPIProcess, NoRankOutput): version: str
[docs] @dataclass class OGSVersionLog2(MPIProcess, NoRankOutput): ogs_version: str log_version: int = 0 log_level: str = ""
[docs] class StepStatus(Enum): NOT_STARTED = "Not started" RUNNING = "Running" TERMINATED = "Terminated" TERMINATED_WITH_ERROR = "Terminated with error" def __str__(self) -> str: return self.value # Ensures printing gives "Running", etc.
[docs] @dataclass class Context: time_step: None | int = None time_step_status: StepStatus = StepStatus.NOT_STARTED process: None | int = None process_step_status: StepStatus = StepStatus.NOT_STARTED iteration_number: None | int = None iteration_step_status: StepStatus = StepStatus.NOT_STARTED simulation_status: StepStatus = StepStatus.NOT_STARTED def __str__(self) -> str: return ( f"Context(\n" f" time_step={self.time_step}, status={self.time_step_status}\n" f" process={self.process}, status={self.process_step_status}\n" f" iteration={self.iteration_number}, status={self.iteration_step_status}\n" f" simulation_status={self.simulation_status}\n" f")" ) def __repr__(self) -> str: return ( f"Context(time_step={self.time_step!r}, time_step_status={self.time_step_status!r}, " f"process={self.process!r}, process_step_status={self.process_step_status!r}, " f"iteration={self.iteration_number!r}, iteration_step_status={self.iteration_step_status!r}, " f"simulation_status={self.simulation_status!r})" )
[docs] def update(self, x: Log | Termination) -> None: if isinstance(x, SimulationStartTime): self.simulation_status = StepStatus.RUNNING if isinstance(x, SimulationEndTime | SimulationExecutionTime): self.simulation_status = StepStatus.TERMINATED if isinstance( x, SimulationEndTimeFailed | SimulationAbort | SimulationExecutionTimeFailed, ): self.simulation_status = StepStatus.TERMINATED_WITH_ERROR if isinstance(x, TimeStepStart): self.time_step = x.time_step self.time_step_status = StepStatus.RUNNING if isinstance(x, TimeStepEnd): assert ( x.time_step == self.time_step ), f"Time step: {x}. Current status: {self.time_step}, {self}" self.time_step_status = StepStatus.TERMINATED if isinstance(x, SolvingProcessStart): assert not self.process or x.process > self.process self.process = x.process self.process_step_status = StepStatus.RUNNING if isinstance(x, SolvingProcessEnd): assert x.process == self.process self.process_step_status = StepStatus.TERMINATED if isinstance(x, IterationStart): self.iteration_number = x.iteration_number self.iteration_step_status = StepStatus.RUNNING if isinstance(x, IterationEnd): assert x.iteration_number == self.iteration_number self.iteration_step_status = StepStatus.TERMINATED
[docs] class TimeStepContext:
[docs] @staticmethod def context_filter() -> list[str]: return ["time_step"]
[docs] class TimeStepProcessContext:
[docs] @staticmethod def context_filter() -> list[str]: return ["time_step", "process"]
[docs] class TimeStepProcessIterationContext:
[docs] @staticmethod def context_filter() -> list[str]: return ["time_step", "process", "iteration_number"]
[docs] @dataclass class AssemblyTime(TimeStepProcessContext, MPIProcess, Info): assembly_time: float
[docs] @dataclass class TimeStepEnd(MPIProcess, Info): time_step: int time_step_finished_time: float
[docs] @dataclass class IterationStart(TimeStepProcessContext, MPIProcess, Info): iteration_number: int
[docs] @dataclass class IterationEnd(TimeStepProcessContext, MPIProcess, Info): iteration_number: int iteration_time: float
[docs] @dataclass class CouplingIterationStart(MPIProcess, Info): coupling_iteration_number: int
[docs] @dataclass class CouplingIterationEnd(MPIProcess, Info): coupling_iteration_number: int coupling_iteration_time: float
[docs] @dataclass class TimeStepStart(MPIProcess, Info): time_step: int step_start_time: float step_size: float
[docs] @dataclass class TimeStepOutputTime(MPIProcess, Info): time_step: int # ToDo from TimeStepContext output_time: float
[docs] @dataclass class SolvingProcessStart(TimeStepContext, MPIProcess, Info): process: int
[docs] @dataclass class SolvingProcessEnd(TimeStepContext, MPIProcess, Info): process: int time_step_solution_time: float time_step: int
[docs] @dataclass class TimeStepSolutionTimeCoupledScheme(MPIProcess, Info): process: int time_step_solution_time: float time_step: int coupling_iteration: int
[docs] @dataclass class TimeStepFinishedTime(MPIProcess, Info): time_step: int time_step_finished_time: float
[docs] @dataclass class DirichletTime(TimeStepProcessContext, MPIProcess, Info): dirichlet_time: float
[docs] @dataclass class LinearSolverTime(TimeStepProcessContext, MPIProcess, Info): linear_solver_time: float
[docs] @dataclass class MeshReadTime(MPIProcess, Info): mesh_read_time: float
[docs] @dataclass class SimulationExecutionTime(MPIProcess, Info, Termination): execution_time: float
[docs] @dataclass class SimulationExecutionTimeFailed(SimulationExecutionTime): pass
[docs] @dataclass class SimulationAbort(Info, Termination): signal: int
[docs] @dataclass class ComponentConvergenceCriterion( TimeStepProcessIterationContext, MPIProcess, Info ): component: int dx: float x: float dx_x: float
[docs] @dataclass class TimeStepConvergenceCriterion( TimeStepProcessIterationContext, MPIProcess, Info ): dx: float x: float dx_x: float
[docs] @dataclass class CouplingIterationConvergence(MPIProcess, Info): coupling_iteration_process: int
[docs] @dataclass class GenericCodePoint(MPIProcess, Info): message: str
[docs] @dataclass class PhaseFieldEnergyVar(MPIProcess, Info): elastic_energy: float surface_energy: float pressure_work: float total_energy: float
[docs] @dataclass class ErrorMessage(MPIProcess, ErrorType): message: str
[docs] @dataclass class CriticalMessage(MPIProcess, CriticalType): message: str
[docs] @dataclass class WarningMessage(MPIProcess, WarningType): message: str
[docs] @dataclass class SimulationStartTime(MPIProcess, Info, NoRankOutput): message: str
[docs] @dataclass class SimulationEndTime(MPIProcess, Info, Termination): message: str
[docs] @dataclass class SimulationEndTimeFailed(MPIProcess, Info, Termination): message: str
[docs] def ogs_regexes() -> list[tuple[str, type[Log]]]: """ Defines regular expressions for parsing OpenGeoSys log messages. :returns: A list of tuples, each containing a regular expression pattern and the corresponding message class. """ return [ ( r"info: This is OpenGeoSys-6 version (\d+)\.(\d+)\.(\d+)(?:-(\d+))?(?:-g([0-9a-f]+))?(?:\.dirty)?", OGSVersionLog, ), ( r"info: \[time\] Output of timestep (\d+) took ([\d\.e+-]+) s", TimeStepOutputTime, ), ( r"info: \[time\] Time step #(\d+) took ([\d\.e+-]+) s", TimeStepFinishedTime, ), (r"info: \[time\] Reading the mesh took ([\d\.e+-]+) s", MeshReadTime), ( r"info: \[time\] Execution took ([\d\.e+-]+) s", SimulationExecutionTime, ), ( r"info: \[time\] Solving process #(\d+) took ([\d\.e+-]+) s in time step #(\d+) coupling iteration #(\d+)", TimeStepSolutionTimeCoupledScheme, ), ( r"info: \[time\] Solving process #(\d+) took ([\d\.e+-]+) s in time step #(\d+)", SolvingProcessEnd, ), ( r"info: === Time stepping at step #(\d+) and time ([\d\.e+-]+) with step size (.*)", TimeStepStart, ), (r"info: \[time\] Assembly took ([\d\.e+-]+) s", AssemblyTime), ( r"info: \[time\] Applying Dirichlet BCs took ([\d\.e+-]+) s", DirichletTime, ), ( r"info: \[time\] Linear solver took ([\d\.e+-]+) s", LinearSolverTime, ), ( r"info: \[time\] Iteration #(\d+) took ([\d\.e+-]+) s", IterationEnd, ), ( r"info: Convergence criterion: \|dx\|=([\d\.e+-]+), \|x\|=([\d\.e+-]+), \|dx\|/\|x\|=([\d\.e+-]+|nan|inf)", TimeStepConvergenceCriterion, ), ( r"info: Elastic energy: ([\d\.e+-]+) Surface energy: ([\d\.e+-]+) Pressure work: ([\d\.e+-]+) Total energy: ([\d\.e+-]+)", PhaseFieldEnergyVar, ), ( r"info: ------- Checking convergence criterion for coupled solution of process #(\d+)", CouplingIterationConvergence, ), ( r"info: ------- Checking convergence criterion for coupled solution of process ID (\d+) -------", CouplingIterationConvergence, ), ( r"info: Convergence criterion, component (\d+): \|dx\|=([\d\.e+-]+), \|x\|=([\d\.e+-]+), \|dx\|/\|x\|=([\d\.e+-]+|nan|inf)$", ComponentConvergenceCriterion, ), ("critical: (.*)", CriticalMessage), ("error: (.*)", ErrorMessage), ("warning: (.*)", WarningMessage), ( r"info: OGS started on (\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}[+-]\d{4}).", SimulationStartTime, ), ( r"info: OGS terminated on (\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}[+-]\d{4}).", SimulationEndTime, ), ]
[docs] def new_regexes() -> list[tuple[str, type[Log]]]: return [ ( r"info: This is OpenGeoSys-6 version: ([\w\-\.]+)\. Log version: (\d+), Log level: (\w+)\.", OGSVersionLog2, ), ( r"info: OGS started on (\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}[+-]\d{4}).", SimulationStartTime, ), ( r"info: Time step #(\d+) started. Time: ([\d\.e+-]+). Step size: (\d+)", TimeStepStart, ), (r"info: \[time\] Reading the mesh took ([\d\.e+-]+) s", MeshReadTime), (r"info: Solving process #(\d+) started", SolvingProcessStart), (r"info: Iteration #(\d+) started", IterationStart), (r"info: \[time\] Assembly took ([\d\.e+-]+) s", AssemblyTime), ( r"info: \[time\] Applying Dirichlet BCs took ([\d\.e+-]+) s", DirichletTime, ), ( r"info: \[time\] Linear solver took ([\d\.e+-]+) s", LinearSolverTime, ), ( r"info: \[time\] Solving process #(\d+) took ([\d\.e+-]+) s in time step #(\d+)", SolvingProcessEnd, ), ( r"info: Convergence criterion, component (\d+): \|dx\|=([\d\.e+-]+), \|x\|=([\d\.e+-]+), \|dx\|/\|x\|=([\d\.e+-]+|nan|inf)$", ComponentConvergenceCriterion, ), ( r"info: Convergence criterion: \|dx\|=([\d\.e+-]+), \|x\|=([\d\.e+-]+), \|dx\|/\|x\|=([\d\.e+-]+|nan|inf)", TimeStepConvergenceCriterion, ), ( r"info: \[time\] Iteration #(\d+) took ([\d\.e+-]+) s", IterationEnd, ), ( r"info: \[time\] Output of timestep (\d+) took ([\d\.e+-]+) s", TimeStepOutputTime, ), ( r"info: \[time\] Time step #(\d+) took ([\d\.e+-]+) s", TimeStepEnd, ), ( r"info: Global coupling iteration #(\d+) started", CouplingIterationStart, ), ( r"info: \[time\] Global coupling iteration #(\d+) took ([\d\.e+-]+) s", CouplingIterationEnd, ), ( r"info: \[time\] Simulation completed. It took ([\d\.e+-]+) s", SimulationExecutionTime, ), ( r"info: \[time\] Simulation failed. It took ([\d\.e+-]+) s", SimulationExecutionTime, ), ( r"info: \[time\] Simulation aborted. Received signal: (\d+).", SimulationAbort, ), ( r"info: OGS terminated on (\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}[+-]\d{4})\.", SimulationEndTime, ), ( r"error: OGS aborted on (\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}[+-]\d{4})\.", SimulationEndTimeFailed, ), ]