Solve and enforce selected ruff PTH rules

* Use pathlib.Path instead of os.path

  These changes include only the fixes ruff was able to safely

* Unsafe-fixes and some manual changes.
* Manual replacement of os.path.join

* Enforce passing ruff PTH rules

Author: berland

pyproject.toml

@@ -141,6 +141,7 @@ select = [
   "PLR",
   "PLW",
   "PT",  # pytest
+  "PTH",
   "PYI", # flake8-pyi
   "Q",
   "RET", # flake8-return
@@ -164,6 +165,9 @@ ignore = [
   "PLR0914", # too-many-local-variables
   "PLR0915", # too-many-statements
   "PLR2004", # magic-value-comparison
+  "PTH118",  # os-path-join
+  "PTH208",  # os-listdir
+  "PTH122",  # os-path-splitext
 ]
 
 [tool.ruff.lint.per-file-ignores]

src/semeio/__init__.py

@@ -1,8 +1,8 @@
 import argparse
 import logging
-import os
 import sys
 from importlib.metadata import PackageNotFoundError, version
+from pathlib import Path
 
 try:  # noqa: SIM105
     __version__ = version(__name__)
@@ -25,6 +25,6 @@
 
 
 def valid_file(arg):
-    if os.path.isfile(arg):
+    if Path(arg).is_file():
         return arg
     raise argparse.ArgumentTypeError(f"{arg} is not an existing file!")

src/semeio/fmudesign/_excel_to_dict.py

@@ -83,7 +83,7 @@ def inputdict_to_yaml(inputdict: dict[str, Any], filename: str) -> None:
         inputdict (dict)
         filename (str): name of output file
     """
-    with open(filename, "w", encoding="utf-8") as stream:
+    with Path(filename).open("w", encoding="utf-8") as stream:
         yaml.dump(inputdict, stream)
 
 

src/semeio/forward_models/design_kw/design_kw.py

@@ -1,8 +1,8 @@
 import logging
-import pathlib
 import re
 import shlex
 from collections.abc import Iterable, Mapping, Sequence
+from pathlib import Path
 from re import Match
 from typing import Any
 
@@ -32,17 +32,17 @@ def run(
 
     valid = True
 
-    with open(parameters_file_name, encoding="utf-8") as parameters_file:
+    with Path(parameters_file_name).open(encoding="utf-8") as parameters_file:
         parameters = parameters_file.readlines()
 
     key_vals = extract_key_value(parameters)
 
     key_vals.update(rm_genkw_prefix(key_vals))
 
-    with open(template_file_name, encoding="utf-8") as template_file:
+    with Path(template_file_name).open(encoding="utf-8") as template_file:
         template = template_file.readlines()
 
-    with open(result_file_name, "w", encoding="utf-8") as result_file:
+    with Path(result_file_name).open("w", encoding="utf-8") as result_file:
         for line in template:
             rendered_line = line
             if not is_comment(rendered_line):
@@ -55,7 +55,7 @@ def run(
             result_file.write(rendered_line)
 
     if valid:
-        pathlib.Path(_STATUS_FILE_NAME).write_text("DESIGN_KW OK\n", encoding="utf-8")
+        Path(_STATUS_FILE_NAME).write_text("DESIGN_KW OK\n", encoding="utf-8")
 
     return valid
 
@@ -202,7 +202,7 @@ def validate_template_keys(
     key_vals: Mapping[str, str], template_file_name: str
 ) -> None:
     try:
-        with open(template_file_name, encoding="utf-8") as template_file:
+        with Path(template_file_name).open(encoding="utf-8") as template_file:
             template = template_file.readlines()
     except (OSError, UnicodeDecodeError) as err:
         ForwardModelStepWarning.warn(
@@ -225,7 +225,7 @@ def validate_configuration(
     template_file_name: str, parameters_file_name: str = "parameters.txt"
 ) -> None:
     try:
-        with open(parameters_file_name, encoding="utf-8") as parameters_file:
+        with Path(parameters_file_name).open(encoding="utf-8") as parameters_file:
             parameters = parameters_file.readlines()
     except (OSError, UnicodeDecodeError) as err:
         ForwardModelStepWarning.warn(

src/semeio/forward_models/overburden_timeshift/ots.py

@@ -32,7 +32,7 @@
 
 def ots_load_params(input_file: str) -> OTSConfig:
     try:
-        with open(input_file, encoding="utf-8") as fin:
+        with Path(input_file).open(encoding="utf-8") as fin:
             config = OTSConfig(**yaml.safe_load(fin))
     except ValidationError as err:
         raise ConfigurationError(
@@ -124,7 +124,7 @@ def ots_run(parameter_file: str) -> None:
             + len(vintage_pairs.ts_rporv)
         )
         line = "{}, {}, {}" + ", {}" * num_pairs + "\n"
-        with open(parms.vintages_export_file, "w", encoding="utf-8") as f:
+        with Path(parms.vintages_export_file).open("w", encoding="utf-8") as f:
             for point, (x_index, y_index) in enumerate(
                 product(
                     range(1, surface_horizon.get_nx() + 1),

src/semeio/forward_models/rft/gendata_rft.py

@@ -2,6 +2,7 @@
 import logging
 import os
 from collections.abc import Mapping, Sequence
+from pathlib import Path
 
 import numpy as np
 import pandas as pd
@@ -74,7 +75,7 @@ def _write_simdata(
     """Write pressure value, one per line for all points, -1 is used where
     there is no pressure information.
     """
-    with open(output_file, "w+", encoding="utf-8") as file_handle:
+    with Path(output_file).open("w+", encoding="utf-8") as file_handle:
         if dataname in trajectory_df:
             file_handle.write(
                 "\n".join(
@@ -93,7 +94,7 @@ def _write_simdata(
 
 def _write_active(output_file: str, trajectory_df: pd.DataFrame) -> None:
     """Write a file with "1" pr row if a point is active, "0" if not"""
-    with open(output_file, "w+", encoding="utf-8") as file_handle:
+    with Path(output_file).open("w+", encoding="utf-8") as file_handle:
         file_handle.write(
             "\n".join(
                 trajectory_df.sort_values("order")["is_active"]
@@ -106,7 +107,7 @@ def _write_active(output_file: str, trajectory_df: pd.DataFrame) -> None:
 
 def _write_inactive_info(output_file: str, trajectory_df: pd.DataFrame) -> None:
     """Write a file with explanations to users for inactive points"""
-    with open(output_file, "w+", encoding="utf-8") as file_handle:
+    with Path(output_file).open("w+", encoding="utf-8") as file_handle:
         if "inactive_info" not in trajectory_df:
             file_handle.write("")
         else:

src/semeio/forward_models/rft/trajectory.py

@@ -1,6 +1,7 @@
 import os
 from collections.abc import Iterator
 from itertools import starmap
+from pathlib import Path
 from typing import Any, Self, TypeAlias
 
 import pandas as pd
@@ -275,10 +276,10 @@ def load_from_file(cls, filepath: str) -> Self:
         trajectory_points: list[Any] = []
 
         filename = os.path.join(filepath)
-        if not os.path.isfile(filename):
+        if not Path(filename).is_file():
             raise OSError(f"Trajectory file {filename} not found!")
 
-        with open(filename, encoding="utf8") as file_handle:
+        with Path(filename).open(encoding="utf8") as file_handle:
             trajectory_lines = file_handle.readlines()
 
         trajectory_lines = [strip_comments(line) for line in trajectory_lines]

src/semeio/forward_models/rft/utility.py

@@ -1,6 +1,5 @@
 import argparse
 import datetime
-import os
 import warnings
 from pathlib import Path
 
@@ -13,7 +12,7 @@ def strip_comments(line: str) -> str:
 
 
 def existing_directory(path: str) -> str:
-    if not os.path.isdir(path):
+    if not Path(path).is_dir():
         raise argparse.ArgumentTypeError(
             f"The path {path} is not an existing directory"
         )
@@ -121,7 +120,7 @@ def valid_eclbase(file_path: str) -> tuple[CornerpointGrid, ResdataRFTFile]:
     Loads both files with respective loaders and returns them
     """
     rft_filepath = file_path + ".RFT"
-    if not os.path.isfile(rft_filepath):
+    if not Path(rft_filepath).is_file():
         raise argparse.ArgumentTypeError(f"The path {rft_filepath} does not exist")
 
     try:
@@ -134,7 +133,7 @@ def valid_eclbase(file_path: str) -> tuple[CornerpointGrid, ResdataRFTFile]:
         ) from err
 
     grid_filepath = file_path + ".EGRID"
-    if not os.path.isfile(grid_filepath):
+    if not Path(grid_filepath).is_file():
         raise argparse.ArgumentTypeError(f"The path {grid_filepath} does not exist")
 
     try:

src/semeio/forward_models/rft/zonemap.py

@@ -1,5 +1,4 @@
 import argparse
-import os
 from collections.abc import Container, Mapping
 from pathlib import Path
 from typing import Any, Self
@@ -34,7 +33,7 @@ def load_and_parse_zonemap_file(cls, filename: str) -> Self | None:
         if filename == "ZONEMAP_NOT_PROVIDED":
             return None
 
-        if not os.path.isfile(filename):
+        if not Path(filename).is_file():
             raise argparse.ArgumentTypeError(f"ZoneMap file {filename} not found!")
 
         zones_at_k_value: dict[int, Any] = {}

src/semeio/forward_models/scripts/fm_pyscal.py

@@ -2,9 +2,9 @@
 
 import argparse
 import logging
-import os
 import sys
 from argparse import ArgumentParser
+from pathlib import Path
 from typing import Literal
 
 from pyscal import pyscalcli
@@ -161,7 +161,7 @@ def run(
     In contrast with the command line tool where interpolation
     parameters are explicit, they are implicit here, and gathered
     from parameters.txt"""
-    if not os.path.exists(relperm_parameters_file):
+    if not Path(relperm_parameters_file).exists():
         _logger.error("%s does not exist", relperm_parameters_file)
         sys.exit(1)
 
@@ -244,9 +244,9 @@ def _get_interpolation_values(
         tuple with two values, one for WaterOil and one for GasOil
     """
     # Read all key-value pairs from parameters.txt
-    if not os.path.exists(parameters_file_name):
+    if not Path(parameters_file_name).exists():
         raise FileNotFoundError(f"{parameters_file_name} does not exist")
-    with open(parameters_file_name, encoding="utf-8") as parameters_file:
+    with Path(parameters_file_name).open(encoding="utf-8") as parameters_file:
         parameters = parameters_file.readlines()
     parameter_dict = extract_key_value(parameters)
     parameter_dict.update(rm_genkw_prefix(parameter_dict))