Skip to content

Replace os.path with pathlib.Path in diffcrash #74

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Apr 11, 2025
Merged

Replace os.path with pathlib.Path in diffcrash #74

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
104 changes: 49 additions & 55 deletions lasso/diffcrash/diffcrash_run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
import typing
from concurrent import futures
from typing import List, Union

from pathlib import Path
import psutil

from ..logging import str_error, str_info, str_running, str_success, str_warn
Expand Down Expand Up @@ -236,10 +236,10 @@ def __init__(

# logdir
if logfile_dir is not None:
self.logfile_dir = logfile_dir
self.logfile_dir = Path(logfile_dir)
else:
self.logfile_dir = os.path.join(project_dir, "Log")
self.logfile_filepath = os.path.join(self.logfile_dir, "DiffcrashRun.log")
self.logfile_dir = Path(project_dir) / "Log"
self.logfile_filepath = self.logfile_dir / "DiffcrashRun.log"

# logger
self.logger = self._setup_logger()
Expand All @@ -248,18 +248,14 @@ def __init__(
self.logger.info(get_application_header())

# diffcrash home
self.diffcrash_home = self._parse_diffcrash_home(diffcrash_home)
self.diffcrash_home = os.path.join(self.diffcrash_home, "bin")
self.diffcrash_lib = os.path.join(os.path.dirname(self.diffcrash_home), "lib")
self.diffcrash_home = Path(self._parse_diffcrash_home(diffcrash_home))
self.diffcrash_home = self.diffcrash_home / "bin"
self.diffcrash_lib = self.diffcrash_home.parent / "lib"

if platform.system() == "Linux":
os.environ["PATH"] = (
os.environ["PATH"] + ":" + self.diffcrash_home + ":" + self.diffcrash_lib
)
os.environ["PATH"] += f":{self.diffcrash_home}:{self.diffcrash_lib}"
if platform.system() == "Windows":
os.environ["PATH"] = (
os.environ["PATH"] + ";" + self.diffcrash_home + ";" + self.diffcrash_lib
)
os.environ["PATH"] += f";{self.diffcrash_home};{self.diffcrash_lib}"

# project dir
self.project_dir = self._parse_project_dir(project_dir)
Expand Down Expand Up @@ -354,7 +350,7 @@ def _parse_crash_code(self, crash_code) -> str:

def _parse_reference_run(self, reference_run) -> str:

reference_run_ok = os.path.isfile(reference_run)
reference_run_ok = Path(reference_run).is_file()

msg = self._msg_option.format("reference-run", reference_run)
print(str_info(msg))
Expand All @@ -376,7 +372,7 @@ def _parse_use_id_mapping(self, use_id_mapping) -> bool:
return use_id_mapping

def _parse_project_dir(self, project_dir):
project_dir = os.path.abspath(project_dir)
project_dir = Path(project_dir).resolve()

msg = self._msg_option.format("project-dir", project_dir)
print(str_info(msg))
Expand All @@ -395,13 +391,13 @@ def _parse_simulation_runs(
simulation_runs = []
for pattern in simulation_run_patterns:
simulation_runs += glob.glob(pattern)
simulation_runs = [filepath for filepath in simulation_runs if os.path.isfile(filepath)]
simulation_runs = [filepath for filepath in simulation_runs if Path(filepath).is_file()]

# search all excluded runs
runs_to_exclude = []
for pattern in exclude_runs:
runs_to_exclude += glob.glob(pattern)
runs_to_exclude = [filepath for filepath in runs_to_exclude if os.path.isfile(filepath)]
runs_to_exclude = [filepath for filepath in runs_to_exclude if Path(filepath).is_file()]

n_runs_before_filtering = len(simulation_runs)
simulation_runs = [
Expand Down Expand Up @@ -448,7 +444,7 @@ def natural_keys(text):
def _parse_config_file(self, config_file) -> Union[str, None]:

_msg_config_file = ""
if len(config_file) > 0 and not os.path.isfile(config_file):
if len(config_file) > 0 and not Path(config_file).is_file():
config_file = None
_msg_config_file = f"Can not find config file '{config_file}'"

Expand All @@ -474,7 +470,7 @@ def _parse_config_file(self, config_file) -> Union[str, None]:
def _parse_parameter_file(self, parameter_file) -> Union[None, str]:

_msg_parameter_file = ""
if len(parameter_file) > 0 and not os.path.isfile(parameter_file):
if len(parameter_file) > 0 and not Path(parameter_file).is_file():
parameter_file = None
_msg_parameter_file = f"Can not find parameter file '{parameter_file}'"
# missing parameter file
Expand Down Expand Up @@ -535,13 +531,13 @@ def run_setup(self, pool: futures.ThreadPoolExecutor):
args = []
if self.config_file is None and self.parameter_file is None:
args = [
os.path.join(self.diffcrash_home, "DFC_Setup_" + self.crash_code + "_fem"),
self.diffcrash_home / f"DFC_Setup_{self.crash_code}_fem",
self.reference_run,
self.project_dir,
]
elif self.config_file is not None and self.parameter_file is None:
args = [
os.path.join(self.diffcrash_home, "DFC_Setup_" + self.crash_code + "_fem"),
self.diffcrash_home / f"DFC_Setup_{self.crash_code}_fem",
self.reference_run,
self.project_dir,
"-C",
Expand All @@ -550,15 +546,15 @@ def run_setup(self, pool: futures.ThreadPoolExecutor):
elif self.config_file is None and self.parameter_file is not None:
if ".fz" in self.reference_run:
args = [
os.path.join(self.diffcrash_home, "DFC_Setup_" + self.crash_code + "_fem"),
self.diffcrash_home / f"DFC_Setup_{self.crash_code}_fem",
self.reference_run,
self.project_dir,
"-P",
self.parameter_file,
]
else:
args = [
os.path.join(self.diffcrash_home, "DFC_Setup_" + self.crash_code),
self.diffcrash_home / f"DFC_Setup_{self.crash_code}",
self.reference_run,
self.project_dir,
"-P",
Expand All @@ -567,7 +563,7 @@ def run_setup(self, pool: futures.ThreadPoolExecutor):
elif self.config_file is not None and self.parameter_file is not None:
if ".fz" in self.reference_run:
args = [
os.path.join(self.diffcrash_home, "DFC_Setup_" + self.crash_code + "_fem"),
self.diffcrash_home / f"DFC_Setup_{self.crash_code}_fem",
self.reference_run,
self.project_dir,
"-C",
Expand All @@ -577,7 +573,7 @@ def run_setup(self, pool: futures.ThreadPoolExecutor):
]
else:
args = [
os.path.join(self.diffcrash_home, "DFC_Setup_" + self.crash_code),
self.diffcrash_home / f"DFC_Setup_{self.crash_code}",
self.reference_run,
self.project_dir,
"-C",
Expand Down Expand Up @@ -649,15 +645,15 @@ def run_import(self, pool: futures.ThreadPoolExecutor):
if self.parameter_file is None:
if self.use_id_mapping:
args = [
os.path.join(self.diffcrash_home, "DFC_Import_" + self.crash_code + "_fem"),
self.diffcrash_home / f"DFC_Import_{self.crash_code}_fem",
"-id",
self.simulation_runs[i_filepath],
self.project_dir,
str(i_filepath + counter_offset),
]
else:
args = [
os.path.join(self.diffcrash_home, "DFC_Import_" + self.crash_code + "_fem"),
self.diffcrash_home / f"DFC_Import_{self.crash_code}_fem",
self.simulation_runs[i_filepath],
self.project_dir,
str(i_filepath + counter_offset),
Expand All @@ -666,15 +662,15 @@ def run_import(self, pool: futures.ThreadPoolExecutor):
else:
if self.use_id_mapping:
args = [
os.path.join(self.diffcrash_home, "DFC_Import_" + self.crash_code),
self.diffcrash_home / f"DFC_Import_{self.crash_code}",
"-ID",
self.simulation_runs[i_filepath],
self.project_dir,
str(i_filepath + counter_offset),
]
else:
args = [
os.path.join(self.diffcrash_home, "DFC_Import_" + self.crash_code),
self.diffcrash_home / f"DFC_Import_{self.crash_code}",
self.simulation_runs[i_filepath],
self.project_dir,
str(i_filepath + counter_offset),
Expand Down Expand Up @@ -779,7 +775,7 @@ def run_math(self, pool: futures.ThreadPoolExecutor):
start_time = time.time()
return_code_future = pool.submit(
run_subprocess,
[os.path.join(self.diffcrash_home, "DFC_Math_" + self.crash_code), self.project_dir],
[self.diffcrash_home / f"DFC_Math_{self.crash_code}", self.project_dir],
)
return_code = return_code_future.result()

Expand Down Expand Up @@ -837,31 +833,31 @@ def run_export(self, pool: futures.ThreadPoolExecutor):
export_item_list = []

# check for pdmx
pdmx_filepath_list = glob.glob(os.path.join(self.project_dir, "*_pdmx"))
pdmx_filepath_list = list(self.project_dir.glob("*_pdmx"))
if pdmx_filepath_list:
export_item_list.append(os.path.basename(pdmx_filepath_list[0]))
export_item_list.append(pdmx_filepath_list[0].name)

# check for pdij
pdij_filepath_list = glob.glob(os.path.join(self.project_dir, "*_pdij"))
pdij_filepath_list = list(self.project_dir.glob("*_pdij"))
if pdij_filepath_list:
export_item_list.append(os.path.basename(pdij_filepath_list[0]))
export_item_list.append(pdij_filepath_list[0].name)

else:
export_item_list = self.read_config_file(self.config_file)

# remove previous existing exports
for export_item in export_item_list:
export_item_filepath = os.path.join(self.project_dir, export_item + ".d3plot.fz")
if os.path.isfile(export_item_filepath):
os.remove(export_item_filepath)
export_item_filepath = self.project_dir / f"{export_item}.d3plot.fz"
if export_item_filepath.is_file():
export_item_filepath.unlink()

# do the thing
start_time = time.time()
return_code_futures = [
pool.submit(
run_subprocess,
[
os.path.join(self.diffcrash_home, "DFC_Export_" + self.crash_code),
self.diffcrash_home / f"DFC_Export_{self.crash_code}",
self.project_dir,
export_item,
],
Expand Down Expand Up @@ -938,7 +934,7 @@ def run_matrix(self, pool: futures.ThreadPoolExecutor):
return_code_future = pool.submit(
run_subprocess,
[
os.path.join(self.diffcrash_home, "DFC_Matrix_" + self.crash_code),
self.diffcrash_home / f"DFC_Matrix_{self.crash_code}",
self.project_dir,
matrix_inputfile,
],
Expand Down Expand Up @@ -1003,7 +999,7 @@ def run_eigen(self, pool: futures.ThreadPoolExecutor):
return_code_future = pool.submit(
run_subprocess,
[
os.path.join(self.diffcrash_home, "DFC_Eigen_" + self.crash_code),
self.diffcrash_home / f"DFC_Eigen_{self.crash_code}",
self.project_dir,
eigen_inputfile,
],
Expand Down Expand Up @@ -1062,16 +1058,16 @@ def run_merge(self, pool: futures.ThreadPoolExecutor):
merge_inputfile = self._create_merge_input_file(self.project_dir)

# clear previous merges
for filepath in glob.glob(os.path.join(self.project_dir, "mode_*")):
if os.path.isfile(filepath):
os.remove(filepath)
for filepath in self.project_dir.glob("mode_*"):
if filepath.is_file():
filepath.unlink()

# run the thing
start_time = time.time()
return_code_future = pool.submit(
run_subprocess,
[
os.path.join(self.diffcrash_home, "DFC_Merge_All_" + self.crash_code),
self.diffcrash_home / f"DFC_Merge_All_{self.crash_code}",
self.project_dir,
merge_inputfile,
],
Expand Down Expand Up @@ -1108,7 +1104,7 @@ def run_merge(self, pool: futures.ThreadPoolExecutor):
print(str_success(msg))
self.logger.info(msg)

def is_logfile_successful(self, logfile: str) -> bool:
def is_logfile_successful(self, logfile: Path) -> bool:
"""Checks if a logfile indicates a success

Parameters
Expand All @@ -1127,7 +1123,7 @@ def is_logfile_successful(self, logfile: str) -> bool:
return True
return False

def _create_merge_input_file(self, directory: str) -> str:
def _create_merge_input_file(self, directory: Path) -> Path:
"""Create an input file for the merge executable

Notes
Expand All @@ -1136,7 +1132,7 @@ def _create_merge_input_file(self, directory: str) -> str:
"""

# creates default inputfile for DFC_Merge
filepath = os.path.join(directory, "merge_all.txt")
filepath = directory / "merge_all.txt"
with open(filepath, "w", encoding="utf-8") as merge_input_file:
merge_input_file.write("eigen_all ! Name of eigen input file\n")
merge_input_file.write(
Expand All @@ -1156,7 +1152,7 @@ def _create_merge_input_file(self, directory: str) -> str:

return filepath

def _create_eigen_input_file(self, directory: str) -> str:
def _create_eigen_input_file(self, directory: Path) -> Path:
"""Create an input file for the eigen executable

Notes
Expand All @@ -1165,7 +1161,7 @@ def _create_eigen_input_file(self, directory: str) -> str:
"""

# creates default inputfile for DFC_Eigen
filepath = os.path.join(directory, "eigen_all.txt")
filepath = directory / "eigen_all.txt"
with open(filepath, "w", encoding="utf-8") as eigen_input_file:
eigen_input_file.write("matrix_all\n")
eigen_input_file.write('""\n')
Expand All @@ -1179,16 +1175,15 @@ def _create_eigen_input_file(self, directory: str) -> str:

return filepath

def _create_matrix_input_file(self, directory: str) -> str:
def _create_matrix_input_file(self, directory: Path) -> Path:
"""Create an input file for the matrix executable

Notes
-----
From the official diffcrash docs.
"""

filepath = directory / "matrix.txt"
# creates default inputfile for DFC_Matrix
filepath = os.path.join(directory, "matrix.txt")
with open(filepath, "w", encoding="utf-8") as matrix_input_file:
matrix_input_file.write("0 1000 ! Initial and final time stept to consider\n")
matrix_input_file.write('"" ! not used\n')
Expand All @@ -1206,7 +1201,7 @@ def clear_project_dir(self):
self.logger.removeHandler(handler)

# delete folder
if os.path.exists(self.project_dir):
if self.project_dir.is_dir():
shutil.rmtree(self.project_dir)

# reinit logger
Expand Down Expand Up @@ -1351,8 +1346,7 @@ def check_if_logfiles_show_success(self, pattern: str) -> List[str]:
_msg_logfile_nok = str_error("Logfile '{0}' reports no success.")
messages = []

logfiles = glob.glob(os.path.join(self.logfile_dir, pattern))
for filepath in logfiles:
for filepath in self.logfile_dir.glob(pattern):
if not self.is_logfile_successful(filepath):
messages.append(_msg_logfile_nok.format(filepath))

Expand Down
2 changes: 1 addition & 1 deletion lasso/femzip/femzip_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,7 +209,7 @@ class FemzipAPI:
_api: Union[None, CDLL] = None

@staticmethod
def load_dynamic_library(path: str) -> CDLL:
def load_dynamic_library(path: Path) -> CDLL:
"""Load a library and check for correct execution

Parameters
Expand Down